remove decorator from external

IPython/config/application.py

@@ -1,621 +1,621 b''
 # encoding: utf-8
 """A base class for a configurable application."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import json
 import logging
 import os
 import re
 import sys
 from copy import deepcopy
 from collections import defaultdict
 
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 from IPython.config.configurable import SingletonConfigurable
 from IPython.config.loader import (
     KVArgParseConfigLoader, PyFileConfigLoader, Config, ArgumentError, ConfigFileNotFound, JSONFileConfigLoader
 )
 
 from IPython.utils.traitlets import (
     Unicode, List, Enum, Dict, Instance, TraitError
 )
 from IPython.utils.importstring import import_item
 from IPython.utils.text import indent, wrap_paragraphs, dedent
 from IPython.utils import py3compat
 from IPython.utils.py3compat import string_types, iteritems
 
 #-----------------------------------------------------------------------------
 # Descriptions for the various sections
 #-----------------------------------------------------------------------------
 
 # merge flags&aliases into options
 option_description = """
 Arguments that take values are actually convenience aliases to full
 Configurables, whose aliases are listed on the help line. For more information
 on full configurables, see '--help-all'.
 """.strip() # trim newlines of front and back
 
 keyvalue_description = """
 Parameters are set from command-line arguments of the form:
 `--Class.trait=value`.
 This line is evaluated in Python, so simple expressions are allowed, e.g.::
 `--C.a='range(3)'` For setting C.a=[0,1,2].
 """.strip() # trim newlines of front and back
 
 # sys.argv can be missing, for example when python is embedded. See the docs
 # for details: http://docs.python.org/2/c-api/intro.html#embedding-python
 if not hasattr(sys, "argv"):
     sys.argv = [""]
 
 subcommand_description = """
 Subcommands are launched as `{app} cmd [args]`. For information on using
 subcommand 'cmd', do: `{app} cmd -h`.
 """
 # get running program name
 
 #-----------------------------------------------------------------------------
 # Application class
 #-----------------------------------------------------------------------------
 
 @decorator
 def catch_config_error(method, app, *args, **kwargs):
     """Method decorator for catching invalid config (Trait/ArgumentErrors) during init.
 
     On a TraitError (generally caused by bad config), this will print the trait's
     message, and exit the app.
     
     For use on init methods, to prevent invoking excepthook on invalid input.
     """
     try:
         return method(app, *args, **kwargs)
     except (TraitError, ArgumentError) as e:
         app.print_help()
         app.log.fatal("Bad config encountered during initialization:")
         app.log.fatal(str(e))
         app.log.debug("Config at the time: %s", app.config)
         app.exit(1)
 
 
 class ApplicationError(Exception):
     pass
 
 class LevelFormatter(logging.Formatter):
     """Formatter with additional `highlevel` record
     
     This field is empty if log level is less than highlevel_limit,
     otherwise it is formatted with self.highlevel_format.
     
     Useful for adding 'WARNING' to warning messages,
     without adding 'INFO' to info, etc.
     """
     highlevel_limit = logging.WARN
     highlevel_format = " %(levelname)s |"
     
     def format(self, record):
         if record.levelno >= self.highlevel_limit:
             record.highlevel = self.highlevel_format % record.__dict__
         else:
             record.highlevel = ""
         return super(LevelFormatter, self).format(record)
             
 
 class Application(SingletonConfigurable):
     """A singleton application with full configuration support."""
 
     # The name of the application, will usually match the name of the command
     # line application
     name = Unicode(u'application')
 
     # The description of the application that is printed at the beginning
     # of the help.
     description = Unicode(u'This is an application.')
     # default section descriptions
     option_description = Unicode(option_description)
     keyvalue_description = Unicode(keyvalue_description)
     subcommand_description = Unicode(subcommand_description)
 
     # The usage and example string that goes at the end of the help string.
     examples = Unicode()
 
     # A sequence of Configurable subclasses whose config=True attributes will
     # be exposed at the command line.
     classes = []
     @property
     def _help_classes(self):
         """Define `App.help_classes` if CLI classes should differ from config file classes"""
         return getattr(self, 'help_classes', self.classes)
     
     @property
     def _config_classes(self):
         """Define `App.config_classes` if config file classes should differ from CLI classes."""
         return getattr(self, 'config_classes', self.classes)
 
     # The version string of this application.
     version = Unicode(u'0.0')
     
     # the argv used to initialize the application
     argv = List()
 
     # The log level for the application
     log_level = Enum((0,10,20,30,40,50,'DEBUG','INFO','WARN','ERROR','CRITICAL'),
                     default_value=logging.WARN,
                     config=True,
                     help="Set the log level by value or name.")
     def _log_level_changed(self, name, old, new):
         """Adjust the log level when log_level is set."""
         if isinstance(new, string_types):
             new = getattr(logging, new)
             self.log_level = new
         self.log.setLevel(new)
     
     _log_formatter_cls = LevelFormatter
     
     log_datefmt = Unicode("%Y-%m-%d %H:%M:%S", config=True,
         help="The date format used by logging formatters for %(asctime)s"
     )
     def _log_datefmt_changed(self, name, old, new):
         self._log_format_changed('log_format', self.log_format, self.log_format)
     
     log_format = Unicode("[%(name)s]%(highlevel)s %(message)s", config=True,
         help="The Logging format template",
     )
     def _log_format_changed(self, name, old, new):
         """Change the log formatter when log_format is set."""
         _log_handler = self.log.handlers[0]
         _log_formatter = self._log_formatter_cls(fmt=new, datefmt=self.log_datefmt)
         _log_handler.setFormatter(_log_formatter)
     
 
     log = Instance(logging.Logger)
     def _log_default(self):
         """Start logging for this application.
 
         The default is to log to stderr using a StreamHandler, if no default
         handler already exists.  The log level starts at logging.WARN, but this
         can be adjusted by setting the ``log_level`` attribute.
         """
         log = logging.getLogger(self.__class__.__name__)
         log.setLevel(self.log_level)
         log.propagate = False
         _log = log # copied from Logger.hasHandlers() (new in Python 3.2)
         while _log:
             if _log.handlers:
                 return log
             if not _log.propagate:
                 break
             else:
                 _log = _log.parent
         if sys.executable.endswith('pythonw.exe'):
             # this should really go to a file, but file-logging is only
             # hooked up in parallel applications
             _log_handler = logging.StreamHandler(open(os.devnull, 'w'))
         else:
             _log_handler = logging.StreamHandler()
         _log_formatter = self._log_formatter_cls(fmt=self.log_format, datefmt=self.log_datefmt)
         _log_handler.setFormatter(_log_formatter)
         log.addHandler(_log_handler)
         return log
 
     # the alias map for configurables
     aliases = Dict({'log-level' : 'Application.log_level'})
 
     # flags for loading Configurables or store_const style flags
     # flags are loaded from this dict by '--key' flags
     # this must be a dict of two-tuples, the first element being the Config/dict
     # and the second being the help string for the flag
     flags = Dict()
     def _flags_changed(self, name, old, new):
         """ensure flags dict is valid"""
         for key,value in iteritems(new):
             assert len(value) == 2, "Bad flag: %r:%s"%(key,value)
             assert isinstance(value[0], (dict, Config)), "Bad flag: %r:%s"%(key,value)
             assert isinstance(value[1], string_types), "Bad flag: %r:%s"%(key,value)
 
 
     # subcommands for launching other applications
     # if this is not empty, this will be a parent Application
     # this must be a dict of two-tuples,
     # the first element being the application class/import string
     # and the second being the help string for the subcommand
     subcommands = Dict()
     # parse_command_line will initialize a subapp, if requested
     subapp = Instance('IPython.config.application.Application', allow_none=True)
 
     # extra command-line arguments that don't set config values
     extra_args = List(Unicode)
 
 
     def __init__(self, **kwargs):
         SingletonConfigurable.__init__(self, **kwargs)
         # Ensure my class is in self.classes, so my attributes appear in command line
         # options and config files.
         if self.__class__ not in self.classes:
             self.classes.insert(0, self.__class__)
 
     def _config_changed(self, name, old, new):
         SingletonConfigurable._config_changed(self, name, old, new)
         self.log.debug('Config changed:')
         self.log.debug(repr(new))
 
     @catch_config_error
     def initialize(self, argv=None):
         """Do the basic steps to configure me.
 
         Override in subclasses.
         """
         self.parse_command_line(argv)
 
 
     def start(self):
         """Start the app mainloop.
 
         Override in subclasses.
         """
         if self.subapp is not None:
             return self.subapp.start()
 
     def print_alias_help(self):
         """Print the alias part of the help."""
         if not self.aliases:
             return
 
         lines = []
         classdict = {}
         for cls in self._help_classes:
             # include all parents (up to, but excluding Configurable) in available names
             for c in cls.mro()[:-3]:
                 classdict[c.__name__] = c
 
         for alias, longname in iteritems(self.aliases):
             classname, traitname = longname.split('.',1)
             cls = classdict[classname]
 
             trait = cls.class_traits(config=True)[traitname]
             help = cls.class_get_trait_help(trait).splitlines()
             # reformat first line
             help[0] = help[0].replace(longname, alias) + ' (%s)'%longname
             if len(alias) == 1:
                 help[0] = help[0].replace('--%s='%alias, '-%s '%alias)
             lines.extend(help)
         # lines.append('')
         print(os.linesep.join(lines))
 
     def print_flag_help(self):
         """Print the flag part of the help."""
         if not self.flags:
             return
 
         lines = []
         for m, (cfg,help) in iteritems(self.flags):
             prefix = '--' if len(m) > 1 else '-'
             lines.append(prefix+m)
             lines.append(indent(dedent(help.strip())))
         # lines.append('')
         print(os.linesep.join(lines))
 
     def print_options(self):
         if not self.flags and not self.aliases:
             return
         lines = ['Options']
         lines.append('-'*len(lines[0]))
         lines.append('')
         for p in wrap_paragraphs(self.option_description):
             lines.append(p)
             lines.append('')
         print(os.linesep.join(lines))
         self.print_flag_help()
         self.print_alias_help()
         print()
 
     def print_subcommands(self):
         """Print the subcommand part of the help."""
         if not self.subcommands:
             return
 
         lines = ["Subcommands"]
         lines.append('-'*len(lines[0]))
         lines.append('')
         for p in wrap_paragraphs(self.subcommand_description.format(
                     app=self.name)):
             lines.append(p)
             lines.append('')
         for subc, (cls, help) in iteritems(self.subcommands):
             lines.append(subc)
             if help:
                 lines.append(indent(dedent(help.strip())))
         lines.append('')
         print(os.linesep.join(lines))
 
     def print_help(self, classes=False):
         """Print the help for each Configurable class in self.classes.
 
         If classes=False (the default), only flags and aliases are printed.
         """
         self.print_description()
         self.print_subcommands()
         self.print_options()
 
         if classes:
             help_classes = self._help_classes
             if help_classes:
                 print("Class parameters")
                 print("----------------")
                 print()
                 for p in wrap_paragraphs(self.keyvalue_description):
                     print(p)
                     print()
 
             for cls in help_classes:
                 cls.class_print_help()
                 print()
         else:
             print("To see all available configurables, use `--help-all`")
             print()
 
         self.print_examples()
 
 
     def print_description(self):
         """Print the application description."""
         for p in wrap_paragraphs(self.description):
             print(p)
             print()
 
     def print_examples(self):
         """Print usage and examples.
 
         This usage string goes at the end of the command line help string
         and should contain examples of the application's usage.
         """
         if self.examples:
             print("Examples")
             print("--------")
             print()
             print(indent(dedent(self.examples.strip())))
             print()
 
     def print_version(self):
         """Print the version string."""
         print(self.version)
 
     def update_config(self, config):
         """Fire the traits events when the config is updated."""
         # Save a copy of the current config.
         newconfig = deepcopy(self.config)
         # Merge the new config into the current one.
         newconfig.merge(config)
         # Save the combined config as self.config, which triggers the traits
         # events.
         self.config = newconfig
 
     @catch_config_error
     def initialize_subcommand(self, subc, argv=None):
         """Initialize a subcommand with argv."""
         subapp,help = self.subcommands.get(subc)
 
         if isinstance(subapp, string_types):
             subapp = import_item(subapp)
 
         # clear existing instances
         self.__class__.clear_instance()
         # instantiate
         self.subapp = subapp.instance(config=self.config)
         # and initialize subapp
         self.subapp.initialize(argv)
     
     def flatten_flags(self):
         """flatten flags and aliases, so cl-args override as expected.
         
         This prevents issues such as an alias pointing to InteractiveShell,
         but a config file setting the same trait in TerminalInteraciveShell
         getting inappropriate priority over the command-line arg.
 
         Only aliases with exactly one descendent in the class list
         will be promoted.
         
         """
         # build a tree of classes in our list that inherit from a particular
         # it will be a dict by parent classname of classes in our list
         # that are descendents
         mro_tree = defaultdict(list)
         for cls in self._help_classes:
             clsname = cls.__name__
             for parent in cls.mro()[1:-3]:
                 # exclude cls itself and Configurable,HasTraits,object
                 mro_tree[parent.__name__].append(clsname)
         # flatten aliases, which have the form:
         # { 'alias' : 'Class.trait' }
         aliases = {}
         for alias, cls_trait in iteritems(self.aliases):
             cls,trait = cls_trait.split('.',1)
             children = mro_tree[cls]
             if len(children) == 1:
                 # exactly one descendent, promote alias
                 cls = children[0]
             aliases[alias] = '.'.join([cls,trait])
         
         # flatten flags, which are of the form:
         # { 'key' : ({'Cls' : {'trait' : value}}, 'help')}
         flags = {}
         for key, (flagdict, help) in iteritems(self.flags):
             newflag = {}
             for cls, subdict in iteritems(flagdict):
                 children = mro_tree[cls]
                 # exactly one descendent, promote flag section
                 if len(children) == 1:
                     cls = children[0]
                 newflag[cls] = subdict
             flags[key] = (newflag, help)
         return flags, aliases
 
     @catch_config_error
     def parse_command_line(self, argv=None):
         """Parse the command line arguments."""
         argv = sys.argv[1:] if argv is None else argv
         self.argv = [ py3compat.cast_unicode(arg) for arg in argv ]
         
         if argv and argv[0] == 'help':
             # turn `ipython help notebook` into `ipython notebook -h`
             argv = argv[1:] + ['-h']
 
         if self.subcommands and len(argv) > 0:
             # we have subcommands, and one may have been specified
             subc, subargv = argv[0], argv[1:]
             if re.match(r'^\w(\-?\w)*$', subc) and subc in self.subcommands:
                 # it's a subcommand, and *not* a flag or class parameter
                 return self.initialize_subcommand(subc, subargv)
 
         # Arguments after a '--' argument are for the script IPython may be
         # about to run, not IPython iteslf. For arguments parsed here (help and
         # version), we want to only search the arguments up to the first
         # occurrence of '--', which we're calling interpreted_argv.
         try:
             interpreted_argv = argv[:argv.index('--')]
         except ValueError:
             interpreted_argv = argv
 
         if any(x in interpreted_argv for x in ('-h', '--help-all', '--help')):
             self.print_help('--help-all' in interpreted_argv)
             self.exit(0)
 
         if '--version' in interpreted_argv or '-V' in interpreted_argv:
             self.print_version()
             self.exit(0)
         
         # flatten flags&aliases, so cl-args get appropriate priority:
         flags,aliases = self.flatten_flags()
         loader = KVArgParseConfigLoader(argv=argv, aliases=aliases,
                                         flags=flags, log=self.log)
         config = loader.load_config()
         self.update_config(config)
         # store unparsed args in extra_args
         self.extra_args = loader.extra_args
 
     @classmethod
     def _load_config_files(cls, basefilename, path=None, log=None):
         """Load config files (py,json) by filename and path.
 
         yield each config object in turn.
         """
         
         if not isinstance(path, list):
             path = [path]
         for path in path[::-1]:
             # path list is in descending priority order, so load files backwards:
             pyloader = PyFileConfigLoader(basefilename+'.py', path=path, log=log)
             jsonloader = JSONFileConfigLoader(basefilename+'.json', path=path, log=log)
             config = None
             for loader in [pyloader, jsonloader]:
                 try:
                     config = loader.load_config()
                 except ConfigFileNotFound:
                     pass
                 except Exception:
                     # try to get the full filename, but it will be empty in the
                     # unlikely event that the error raised before filefind finished
                     filename = loader.full_filename or basefilename
                     # problem while running the file
                     if log:
                         log.error("Exception while loading config file %s",
                                 filename, exc_info=True)
                 else:
                     if log:
                         log.debug("Loaded config file: %s", loader.full_filename)
                 if config:
                      yield config
 
         raise StopIteration
 
 
     @catch_config_error
     def load_config_file(self, filename, path=None):
         """Load config files by filename and path."""
         filename, ext = os.path.splitext(filename)
         loaded = []
         for config in self._load_config_files(filename, path=path, log=self.log):
             loaded.append(config)
             self.update_config(config)
         if len(loaded) > 1:
             collisions = loaded[0].collisions(loaded[1])
             if collisions:
                 self.log.warn("Collisions detected in {0}.py and {0}.json config files."
                               " {0}.json has higher priority: {1}".format(
                               filename, json.dumps(collisions, indent=2),
                 ))
 
 
     def generate_config_file(self):
         """generate default config file from Configurables"""
         lines = ["# Configuration file for %s."%self.name]
         lines.append('')
         lines.append('c = get_config()')
         lines.append('')
         for cls in self._config_classes:
             lines.append(cls.class_config_section())
         return '\n'.join(lines)
 
     def exit(self, exit_status=0):
         self.log.debug("Exiting application: %s" % self.name)
         sys.exit(exit_status)
 
     @classmethod
     def launch_instance(cls, argv=None, **kwargs):
         """Launch a global instance of this Application
         
         If a global instance already exists, this reinitializes and starts it
         """
         app = cls.instance(**kwargs)
         app.initialize(argv)
         app.start()
 
 #-----------------------------------------------------------------------------
 # utility functions, for convenience
 #-----------------------------------------------------------------------------
 
 def boolean_flag(name, configurable, set_help='', unset_help=''):
     """Helper for building basic --trait, --no-trait flags.
 
     Parameters
     ----------
 
     name : str
         The name of the flag.
     configurable : str
         The 'Class.trait' string of the trait to be set/unset with the flag
     set_help : unicode
         help string for --name flag
     unset_help : unicode
         help string for --no-name flag
 
     Returns
     -------
 
     cfg : dict
         A dict with two keys: 'name', and 'no-name', for setting and unsetting
         the trait, respectively.
     """
     # default helpstrings
     set_help = set_help or "set %s=True"%configurable
     unset_help = unset_help or "set %s=False"%configurable
 
     cls,trait = configurable.split('.')
 
     setter = {cls : {trait : True}}
     unsetter = {cls : {trait : False}}
     return {name : (setter, set_help), 'no-'+name : (unsetter, unset_help)}
 
 
 def get_config():
     """Get the config object for the global Application instance, if there is one
     
     otherwise return an empty config object
     """
     if Application.initialized():
         return Application.instance().config
     else:
         return Config()

IPython/core/formatters.py

@@ -1,965 +1,965 b''
 # -*- coding: utf-8 -*-
 """Display formatters.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.core.formatters
    :parts: 3
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import abc
 import inspect
 import json
 import sys
 import traceback
 import warnings
 
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 from IPython.config.configurable import Configurable
 from IPython.core.getipython import get_ipython
 from IPython.lib import pretty
 from IPython.utils.traitlets import (
     Bool, Dict, Integer, Unicode, CUnicode, ObjectName, List,
     ForwardDeclaredInstance,
 )
 from IPython.utils.py3compat import (
     with_metaclass, string_types, unicode_type,
 )
 
 
 #-----------------------------------------------------------------------------
 # The main DisplayFormatter class
 #-----------------------------------------------------------------------------
 
 
 def _safe_get_formatter_method(obj, name):
     """Safely get a formatter method
     
     - Classes cannot have formatter methods, only instance
     - protect against proxy objects that claim to have everything
     """
     if inspect.isclass(obj):
         # repr methods only make sense on instances, not classes
         return None
     method = pretty._safe_getattr(obj, name, None)
     if callable(method):
         # obj claims to have repr method...
         if callable(pretty._safe_getattr(obj, '_ipython_canary_method_should_not_exist_', None)):
             # ...but don't trust proxy objects that claim to have everything
             return None
         return method
 
 
 class DisplayFormatter(Configurable):
 
     # When set to true only the default plain text formatter will be used.
     plain_text_only = Bool(False, config=True)
     def _plain_text_only_changed(self, name, old, new):
         warnings.warn("""DisplayFormatter.plain_text_only is deprecated.
         
         Use DisplayFormatter.active_types = ['text/plain']
         for the same effect.
         """, DeprecationWarning)
         if new:
             self.active_types = ['text/plain']
         else:
             self.active_types = self.format_types
     
     active_types = List(Unicode, config=True,
         help="""List of currently active mime-types to display.
         You can use this to set a white-list for formats to display.
         
         Most users will not need to change this value.
         """)
     def _active_types_default(self):
         return self.format_types
     
     def _active_types_changed(self, name, old, new):
         for key, formatter in self.formatters.items():
             if key in new:
                 formatter.enabled = True
             else:
                 formatter.enabled = False
     
     ipython_display_formatter = ForwardDeclaredInstance('FormatterABC')
     def _ipython_display_formatter_default(self):
         return IPythonDisplayFormatter(parent=self)
     
     # A dict of formatter whose keys are format types (MIME types) and whose
     # values are subclasses of BaseFormatter.
     formatters = Dict()
     def _formatters_default(self):
         """Activate the default formatters."""
         formatter_classes = [
             PlainTextFormatter,
             HTMLFormatter,
             MarkdownFormatter,
             SVGFormatter,
             PNGFormatter,
             PDFFormatter,
             JPEGFormatter,
             LatexFormatter,
             JSONFormatter,
             JavascriptFormatter
         ]
         d = {}
         for cls in formatter_classes:
             f = cls(parent=self)
             d[f.format_type] = f
         return d
 
     def format(self, obj, include=None, exclude=None):
         """Return a format data dict for an object.
 
         By default all format types will be computed.
 
         The following MIME types are currently implemented:
 
         * text/plain
         * text/html
         * text/markdown
         * text/latex
         * application/json
         * application/javascript
         * application/pdf
         * image/png
         * image/jpeg
         * image/svg+xml
 
         Parameters
         ----------
         obj : object
             The Python object whose format data will be computed.
         include : list or tuple, optional
             A list of format type strings (MIME types) to include in the
             format data dict. If this is set *only* the format types included
             in this list will be computed.
         exclude : list or tuple, optional
             A list of format type string (MIME types) to exclude in the format
             data dict. If this is set all format types will be computed,
             except for those included in this argument.
 
         Returns
         -------
         (format_dict, metadata_dict) : tuple of two dicts
         
             format_dict is a dictionary of key/value pairs, one of each format that was
             generated for the object. The keys are the format types, which
             will usually be MIME type strings and the values and JSON'able
             data structure containing the raw data for the representation in
             that format.
             
             metadata_dict is a dictionary of metadata about each mime-type output.
             Its keys will be a strict subset of the keys in format_dict.
         """
         format_dict = {}
         md_dict = {}
         
         if self.ipython_display_formatter(obj):
             # object handled itself, don't proceed
             return {}, {}
         
         for format_type, formatter in self.formatters.items():
             if include and format_type not in include:
                 continue
             if exclude and format_type in exclude:
                 continue
             
             md = None
             try:
                 data = formatter(obj)
             except:
                 # FIXME: log the exception
                 raise
             
             # formatters can return raw data or (data, metadata)
             if isinstance(data, tuple) and len(data) == 2:
                 data, md = data
             
             if data is not None:
                 format_dict[format_type] = data
             if md is not None:
                 md_dict[format_type] = md
             
         return format_dict, md_dict
 
     @property
     def format_types(self):
         """Return the format types (MIME types) of the active formatters."""
         return list(self.formatters.keys())
 
 
 #-----------------------------------------------------------------------------
 # Formatters for specific format types (text, html, svg, etc.)
 #-----------------------------------------------------------------------------
 
 
 def _safe_repr(obj):
     """Try to return a repr of an object
 
     always returns a string, at least.
     """
     try:
         return repr(obj)
     except Exception as e:
         return "un-repr-able object (%r)" % e
 
 
 class FormatterWarning(UserWarning):
     """Warning class for errors in formatters"""
 
 @decorator
 def catch_format_error(method, self, *args, **kwargs):
     """show traceback on failed format call"""
     try:
         r = method(self, *args, **kwargs)
     except NotImplementedError:
         # don't warn on NotImplementedErrors
         return None
     except Exception:
         exc_info = sys.exc_info()
         ip = get_ipython()
         if ip is not None:
             ip.showtraceback(exc_info)
         else:
             traceback.print_exception(*exc_info)
         return None
     return self._check_return(r, args[0])
 
 
 class FormatterABC(with_metaclass(abc.ABCMeta, object)):
     """ Abstract base class for Formatters.
 
     A formatter is a callable class that is responsible for computing the
     raw format data for a particular format type (MIME type). For example,
     an HTML formatter would have a format type of `text/html` and would return
     the HTML representation of the object when called.
     """
 
     # The format type of the data returned, usually a MIME type.
     format_type = 'text/plain'
 
     # Is the formatter enabled...
     enabled = True
     
     @abc.abstractmethod
     def __call__(self, obj):
         """Return a JSON'able representation of the object.
 
         If the object cannot be formatted by this formatter,
         warn and return None.
         """
         return repr(obj)
 
 
 def _mod_name_key(typ):
     """Return a (__module__, __name__) tuple for a type.
     
     Used as key in Formatter.deferred_printers.
     """
     module = getattr(typ, '__module__', None)
     name = getattr(typ, '__name__', None)
     return (module, name)
 
 
 def _get_type(obj):
     """Return the type of an instance (old and new-style)"""
     return getattr(obj, '__class__', None) or type(obj)
 
 _raise_key_error = object()
 
 
 class BaseFormatter(Configurable):
     """A base formatter class that is configurable.
 
     This formatter should usually be used as the base class of all formatters.
     It is a traited :class:`Configurable` class and includes an extensible
     API for users to determine how their objects are formatted. The following
     logic is used to find a function to format an given object.
 
     1. The object is introspected to see if it has a method with the name
        :attr:`print_method`. If is does, that object is passed to that method
        for formatting.
     2. If no print method is found, three internal dictionaries are consulted
        to find print method: :attr:`singleton_printers`, :attr:`type_printers`
        and :attr:`deferred_printers`.
 
     Users should use these dictionaries to register functions that will be
     used to compute the format data for their objects (if those objects don't
     have the special print methods). The easiest way of using these
     dictionaries is through the :meth:`for_type` and :meth:`for_type_by_name`
     methods.
 
     If no function/callable is found to compute the format data, ``None`` is
     returned and this format type is not used.
     """
 
     format_type = Unicode('text/plain')
     _return_type = string_types
 
     enabled = Bool(True, config=True)
 
     print_method = ObjectName('__repr__')
 
     # The singleton printers.
     # Maps the IDs of the builtin singleton objects to the format functions.
     singleton_printers = Dict(config=True)
 
     # The type-specific printers.
     # Map type objects to the format functions.
     type_printers = Dict(config=True)
 
     # The deferred-import type-specific printers.
     # Map (modulename, classname) pairs to the format functions.
     deferred_printers = Dict(config=True)
     
     @catch_format_error
     def __call__(self, obj):
         """Compute the format for an object."""
         if self.enabled:
             # lookup registered printer
             try:
                 printer = self.lookup(obj)
             except KeyError:
                 pass
             else:
                 return printer(obj)
             # Finally look for special method names
             method = _safe_get_formatter_method(obj, self.print_method)
             if method is not None:
                 return method()
             return None
         else:
             return None
     
     def __contains__(self, typ):
         """map in to lookup_by_type"""
         try:
             self.lookup_by_type(typ)
         except KeyError:
             return False
         else:
             return True
     
     def _check_return(self, r, obj):
         """Check that a return value is appropriate
         
         Return the value if so, None otherwise, warning if invalid.
         """
         if r is None or isinstance(r, self._return_type) or \
             (isinstance(r, tuple) and r and isinstance(r[0], self._return_type)):
             return r
         else:
             warnings.warn(
                 "%s formatter returned invalid type %s (expected %s) for object: %s" % \
                 (self.format_type, type(r), self._return_type, _safe_repr(obj)),
                 FormatterWarning
             )
     
     def lookup(self, obj):
         """Look up the formatter for a given instance.
         
         Parameters
         ----------
         obj  : object instance
 
         Returns
         -------
         f : callable
             The registered formatting callable for the type.
 
         Raises
         ------
         KeyError if the type has not been registered.
         """
         # look for singleton first
         obj_id = id(obj)
         if obj_id in self.singleton_printers:
             return self.singleton_printers[obj_id]
         # then lookup by type
         return self.lookup_by_type(_get_type(obj))
     
     def lookup_by_type(self, typ):
         """Look up the registered formatter for a type.
 
         Parameters
         ----------
         typ  : type or '__module__.__name__' string for a type
 
         Returns
         -------
         f : callable
             The registered formatting callable for the type.
 
         Raises
         ------
         KeyError if the type has not been registered.
         """
         if isinstance(typ, string_types):
             typ_key = tuple(typ.rsplit('.',1))
             if typ_key not in self.deferred_printers:
                 # We may have it cached in the type map. We will have to
                 # iterate over all of the types to check.
                 for cls in self.type_printers:
                     if _mod_name_key(cls) == typ_key:
                         return self.type_printers[cls]
             else:
                 return self.deferred_printers[typ_key]
         else:
             for cls in pretty._get_mro(typ):
                 if cls in self.type_printers or self._in_deferred_types(cls):
                     return self.type_printers[cls]
         
         # If we have reached here, the lookup failed.
         raise KeyError("No registered printer for {0!r}".format(typ))
 
     def for_type(self, typ, func=None):
         """Add a format function for a given type.
         
         Parameters
         -----------
         typ : type or '__module__.__name__' string for a type
             The class of the object that will be formatted using `func`.
         func : callable
             A callable for computing the format data.
             `func` will be called with the object to be formatted,
             and will return the raw data in this formatter's format.
             Subclasses may use a different call signature for the
             `func` argument.
             
             If `func` is None or not specified, there will be no change,
             only returning the current value.
         
         Returns
         -------
         oldfunc : callable
             The currently registered callable.
             If you are registering a new formatter,
             this will be the previous value (to enable restoring later).
         """
         # if string given, interpret as 'pkg.module.class_name'
         if isinstance(typ, string_types):
             type_module, type_name = typ.rsplit('.', 1)
             return self.for_type_by_name(type_module, type_name, func)
         
         try:
             oldfunc = self.lookup_by_type(typ)
         except KeyError:
             oldfunc = None
         
         if func is not None:
             self.type_printers[typ] = func
         
         return oldfunc
 
     def for_type_by_name(self, type_module, type_name, func=None):
         """Add a format function for a type specified by the full dotted
         module and name of the type, rather than the type of the object.
 
         Parameters
         ----------
         type_module : str
             The full dotted name of the module the type is defined in, like
             ``numpy``.
         type_name : str
             The name of the type (the class name), like ``dtype``
         func : callable
             A callable for computing the format data.
             `func` will be called with the object to be formatted,
             and will return the raw data in this formatter's format.
             Subclasses may use a different call signature for the
             `func` argument.
             
             If `func` is None or unspecified, there will be no change,
             only returning the current value.
         
         Returns
         -------
         oldfunc : callable
             The currently registered callable.
             If you are registering a new formatter,
             this will be the previous value (to enable restoring later).
         """
         key = (type_module, type_name)
         
         try:
             oldfunc = self.lookup_by_type("%s.%s" % key)
         except KeyError:
             oldfunc = None
         
         if func is not None:
             self.deferred_printers[key] = func
         return oldfunc
     
     def pop(self, typ, default=_raise_key_error):
         """Pop a formatter for the given type.
 
         Parameters
         ----------
         typ : type or '__module__.__name__' string for a type
         default : object
             value to be returned if no formatter is registered for typ.
 
         Returns
         -------
         obj : object
             The last registered object for the type.
 
         Raises
         ------
         KeyError if the type is not registered and default is not specified.
         """
         
         if isinstance(typ, string_types):
             typ_key = tuple(typ.rsplit('.',1))
             if typ_key not in self.deferred_printers:
                 # We may have it cached in the type map. We will have to
                 # iterate over all of the types to check.
                 for cls in self.type_printers:
                     if _mod_name_key(cls) == typ_key:
                         old = self.type_printers.pop(cls)
                         break
                 else:
                     old = default
             else:
                 old = self.deferred_printers.pop(typ_key)
         else:
             if typ in self.type_printers:
                 old = self.type_printers.pop(typ)
             else:
                 old = self.deferred_printers.pop(_mod_name_key(typ), default)
         if old is _raise_key_error:
             raise KeyError("No registered value for {0!r}".format(typ))
         return old
 
     def _in_deferred_types(self, cls):
         """
         Check if the given class is specified in the deferred type registry.
 
         Successful matches will be moved to the regular type registry for future use.
         """
         mod = getattr(cls, '__module__', None)
         name = getattr(cls, '__name__', None)
         key = (mod, name)
         if key in self.deferred_printers:
             # Move the printer over to the regular registry.
             printer = self.deferred_printers.pop(key)
             self.type_printers[cls] = printer
             return True
         return False
 
 
 class PlainTextFormatter(BaseFormatter):
     """The default pretty-printer.
 
     This uses :mod:`IPython.lib.pretty` to compute the format data of
     the object. If the object cannot be pretty printed, :func:`repr` is used.
     See the documentation of :mod:`IPython.lib.pretty` for details on
     how to write pretty printers.  Here is a simple example::
 
         def dtype_pprinter(obj, p, cycle):
             if cycle:
                 return p.text('dtype(...)')
             if hasattr(obj, 'fields'):
                 if obj.fields is None:
                     p.text(repr(obj))
                 else:
                     p.begin_group(7, 'dtype([')
                     for i, field in enumerate(obj.descr):
                         if i > 0:
                             p.text(',')
                             p.breakable()
                         p.pretty(field)
                     p.end_group(7, '])')
     """
 
     # The format type of data returned.
     format_type = Unicode('text/plain')
 
     # This subclass ignores this attribute as it always need to return
     # something.
     enabled = Bool(True, config=False)
     
     max_seq_length = Integer(pretty.MAX_SEQ_LENGTH, config=True,
         help="""Truncate large collections (lists, dicts, tuples, sets) to this size.
         
         Set to 0 to disable truncation.
         """
     )
     
     # Look for a _repr_pretty_ methods to use for pretty printing.
     print_method = ObjectName('_repr_pretty_')
 
     # Whether to pretty-print or not.
     pprint = Bool(True, config=True)
 
     # Whether to be verbose or not.
     verbose = Bool(False, config=True)
 
     # The maximum width.
     max_width = Integer(79, config=True)
 
     # The newline character.
     newline = Unicode('\n', config=True)
 
     # format-string for pprinting floats
     float_format = Unicode('%r')
     # setter for float precision, either int or direct format-string
     float_precision = CUnicode('', config=True)
 
     def _float_precision_changed(self, name, old, new):
         """float_precision changed, set float_format accordingly.
 
         float_precision can be set by int or str.
         This will set float_format, after interpreting input.
         If numpy has been imported, numpy print precision will also be set.
 
         integer `n` sets format to '%.nf', otherwise, format set directly.
 
         An empty string returns to defaults (repr for float, 8 for numpy).
 
         This parameter can be set via the '%precision' magic.
         """
 
         if '%' in new:
             # got explicit format string
             fmt = new
             try:
                 fmt%3.14159
             except Exception:
                 raise ValueError("Precision must be int or format string, not %r"%new)
         elif new:
             # otherwise, should be an int
             try:
                 i = int(new)
                 assert i >= 0
             except ValueError:
                 raise ValueError("Precision must be int or format string, not %r"%new)
             except AssertionError:
                 raise ValueError("int precision must be non-negative, not %r"%i)
 
             fmt = '%%.%if'%i
             if 'numpy' in sys.modules:
                 # set numpy precision if it has been imported
                 import numpy
                 numpy.set_printoptions(precision=i)
         else:
             # default back to repr
             fmt = '%r'
             if 'numpy' in sys.modules:
                 import numpy
                 # numpy default is 8
                 numpy.set_printoptions(precision=8)
         self.float_format = fmt
 
     # Use the default pretty printers from IPython.lib.pretty.
     def _singleton_printers_default(self):
         return pretty._singleton_pprinters.copy()
 
     def _type_printers_default(self):
         d = pretty._type_pprinters.copy()
         d[float] = lambda obj,p,cycle: p.text(self.float_format%obj)
         return d
 
     def _deferred_printers_default(self):
         return pretty._deferred_type_pprinters.copy()
 
     #### FormatterABC interface ####
 
     @catch_format_error
     def __call__(self, obj):
         """Compute the pretty representation of the object."""
         if not self.pprint:
             return repr(obj)
         else:
             # handle str and unicode on Python 2
             # io.StringIO only accepts unicode,
             # cStringIO doesn't handle unicode on py2,
             # StringIO allows str, unicode but only ascii str
             stream = pretty.CUnicodeIO()
             printer = pretty.RepresentationPrinter(stream, self.verbose,
                 self.max_width, self.newline,
                 max_seq_length=self.max_seq_length,
                 singleton_pprinters=self.singleton_printers,
                 type_pprinters=self.type_printers,
                 deferred_pprinters=self.deferred_printers)
             printer.pretty(obj)
             printer.flush()
             return stream.getvalue()
 
 
 class HTMLFormatter(BaseFormatter):
     """An HTML formatter.
 
     To define the callables that compute the HTML representation of your
     objects, define a :meth:`_repr_html_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a valid HTML snippet that
     could be injected into an existing DOM. It should *not* include the
     ```<html>`` or ```<body>`` tags.
     """
     format_type = Unicode('text/html')
 
     print_method = ObjectName('_repr_html_')
 
 
 class MarkdownFormatter(BaseFormatter):
     """A Markdown formatter.
 
     To define the callables that compute the Markdown representation of your
     objects, define a :meth:`_repr_markdown_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a valid Markdown.
     """
     format_type = Unicode('text/markdown')
 
     print_method = ObjectName('_repr_markdown_')
 
 class SVGFormatter(BaseFormatter):
     """An SVG formatter.
 
     To define the callables that compute the SVG representation of your
     objects, define a :meth:`_repr_svg_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be valid SVG enclosed in
     ```<svg>``` tags, that could be injected into an existing DOM. It should
     *not* include the ```<html>`` or ```<body>`` tags.
     """
     format_type = Unicode('image/svg+xml')
 
     print_method = ObjectName('_repr_svg_')
 
 
 class PNGFormatter(BaseFormatter):
     """A PNG formatter.
 
     To define the callables that compute the PNG representation of your
     objects, define a :meth:`_repr_png_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be raw PNG data, *not*
     base64 encoded.
     """
     format_type = Unicode('image/png')
 
     print_method = ObjectName('_repr_png_')
     
     _return_type = (bytes, unicode_type)
 
 
 class JPEGFormatter(BaseFormatter):
     """A JPEG formatter.
 
     To define the callables that compute the JPEG representation of your
     objects, define a :meth:`_repr_jpeg_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be raw JPEG data, *not*
     base64 encoded.
     """
     format_type = Unicode('image/jpeg')
 
     print_method = ObjectName('_repr_jpeg_')
 
     _return_type = (bytes, unicode_type)
 
 
 class LatexFormatter(BaseFormatter):
     """A LaTeX formatter.
 
     To define the callables that compute the LaTeX representation of your
     objects, define a :meth:`_repr_latex_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a valid LaTeX equation,
     enclosed in either ```$```, ```$$``` or another LaTeX equation
     environment.
     """
     format_type = Unicode('text/latex')
 
     print_method = ObjectName('_repr_latex_')
 
 
 class JSONFormatter(BaseFormatter):
     """A JSON string formatter.
 
     To define the callables that compute the JSONable representation of
     your objects, define a :meth:`_repr_json_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be a JSONable list or dict.
     JSON scalars (None, number, string) are not allowed, only dict or list containers.
     """
     format_type = Unicode('application/json')
     _return_type = (list, dict)
 
     print_method = ObjectName('_repr_json_')
     
     def _check_return(self, r, obj):
         """Check that a return value is appropriate
         
         Return the value if so, None otherwise, warning if invalid.
         """
         if r is None:
             return
         md = None
         if isinstance(r, tuple):
             # unpack data, metadata tuple for type checking on first element
             r, md = r
         
         # handle deprecated JSON-as-string form from IPython < 3
         if isinstance(r, string_types):
             warnings.warn("JSON expects JSONable list/dict containers, not JSON strings",
             FormatterWarning)
             r = json.loads(r)
         
         if md is not None:
             # put the tuple back together
             r = (r, md)
         return super(JSONFormatter, self)._check_return(r, obj)
 
 
 class JavascriptFormatter(BaseFormatter):
     """A Javascript formatter.
 
     To define the callables that compute the Javascript representation of
     your objects, define a :meth:`_repr_javascript_` method or use the
     :meth:`for_type` or :meth:`for_type_by_name` methods to register functions
     that handle this.
 
     The return value of this formatter should be valid Javascript code and
     should *not* be enclosed in ```<script>``` tags.
     """
     format_type = Unicode('application/javascript')
 
     print_method = ObjectName('_repr_javascript_')
 
 
 class PDFFormatter(BaseFormatter):
     """A PDF formatter.
 
     To define the callables that compute the PDF representation of your
     objects, define a :meth:`_repr_pdf_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this.
 
     The return value of this formatter should be raw PDF data, *not*
     base64 encoded.
     """
     format_type = Unicode('application/pdf')
 
     print_method = ObjectName('_repr_pdf_')
 
     _return_type = (bytes, unicode_type)
 
 class IPythonDisplayFormatter(BaseFormatter):
     """A Formatter for objects that know how to display themselves.
     
     To define the callables that compute the representation of your
     objects, define a :meth:`_ipython_display_` method or use the :meth:`for_type`
     or :meth:`for_type_by_name` methods to register functions that handle
     this. Unlike mime-type displays, this method should not return anything,
     instead calling any appropriate display methods itself.
     
     This display formatter has highest priority.
     If it fires, no other display formatter will be called.
     """
     print_method = ObjectName('_ipython_display_')
     _return_type = (type(None), bool)
     
 
     @catch_format_error
     def __call__(self, obj):
         """Compute the format for an object."""
         if self.enabled:
             # lookup registered printer
             try:
                 printer = self.lookup(obj)
             except KeyError:
                 pass
             else:
                 printer(obj)
                 return True
             # Finally look for special method names
             method = _safe_get_formatter_method(obj, self.print_method)
             if method is not None:
                 method()
                 return True
 
 
 FormatterABC.register(BaseFormatter)
 FormatterABC.register(PlainTextFormatter)
 FormatterABC.register(HTMLFormatter)
 FormatterABC.register(MarkdownFormatter)
 FormatterABC.register(SVGFormatter)
 FormatterABC.register(PNGFormatter)
 FormatterABC.register(PDFFormatter)
 FormatterABC.register(JPEGFormatter)
 FormatterABC.register(LatexFormatter)
 FormatterABC.register(JSONFormatter)
 FormatterABC.register(JavascriptFormatter)
 FormatterABC.register(IPythonDisplayFormatter)
 
 
 def format_display_data(obj, include=None, exclude=None):
     """Return a format data dict for an object.
 
     By default all format types will be computed.
 
     The following MIME types are currently implemented:
 
     * text/plain
     * text/html
     * text/markdown
     * text/latex
     * application/json
     * application/javascript
     * application/pdf
     * image/png
     * image/jpeg
     * image/svg+xml
 
     Parameters
     ----------
     obj : object
         The Python object whose format data will be computed.
 
     Returns
     -------
     format_dict : dict
         A dictionary of key/value pairs, one or each format that was
         generated for the object. The keys are the format types, which
         will usually be MIME type strings and the values and JSON'able
         data structure containing the raw data for the representation in
         that format.
     include : list or tuple, optional
         A list of format type strings (MIME types) to include in the
         format data dict. If this is set *only* the format types included
         in this list will be computed.
     exclude : list or tuple, optional
         A list of format type string (MIME types) to exclue in the format
         data dict. If this is set all format types will be computed,
         except for those included in this argument.
     """
     from IPython.core.interactiveshell import InteractiveShell
 
     InteractiveShell.instance().display_formatter.format(
         obj,
         include,
         exclude
     )
 

IPython/core/history.py

@@ -1,870 +1,870 b''
 """ History related magics and functionality """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011 The IPython Development Team.
 #
 #  Distributed under the terms of the BSD License.
 #
 #  The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib imports
 import atexit
 import datetime
 import os
 import re
 try:
     import sqlite3
 except ImportError:
     try:
         from pysqlite2 import dbapi2 as sqlite3
     except ImportError:
         sqlite3 = None
 import threading
 
 # Our own packages
 from IPython.config.configurable import Configurable
-from IPython.external.decorator import decorator
+from decorator import decorator
 from IPython.utils.decorators import undoc
 from IPython.utils.path import locate_profile
 from IPython.utils import py3compat
 from IPython.utils.traitlets import (
     Any, Bool, Dict, Instance, Integer, List, Unicode, TraitError,
 )
 from IPython.utils.warn import warn
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 @undoc
 class DummyDB(object):
     """Dummy DB that will act as a black hole for history.
     
     Only used in the absence of sqlite"""
     def execute(*args, **kwargs):
         return []
     
     def commit(self, *args, **kwargs):
         pass
     
     def __enter__(self, *args, **kwargs):
         pass
     
     def __exit__(self, *args, **kwargs):
         pass
 
 
 @decorator
 def needs_sqlite(f, self, *a, **kw):
     """Decorator: return an empty list in the absence of sqlite."""
     if sqlite3 is None or not self.enabled:
         return []
     else:
         return f(self, *a, **kw)
 
 
 if sqlite3 is not None:
     DatabaseError = sqlite3.DatabaseError
 else:
     @undoc
     class DatabaseError(Exception):
         "Dummy exception when sqlite could not be imported. Should never occur."
 
 @decorator
 def catch_corrupt_db(f, self, *a, **kw):
     """A decorator which wraps HistoryAccessor method calls to catch errors from
     a corrupt SQLite database, move the old database out of the way, and create
     a new one.
     """
     try:
         return f(self, *a, **kw)
     except DatabaseError:
         if os.path.isfile(self.hist_file):
             # Try to move the file out of the way
             base,ext = os.path.splitext(self.hist_file)
             newpath = base + '-corrupt' + ext
             os.rename(self.hist_file, newpath)
             self.init_db()
             print("ERROR! History file wasn't a valid SQLite database.",
             "It was moved to %s" % newpath, "and a new file created.")
             return []
         
         else:
             # The hist_file is probably :memory: or something else.
             raise
         
 class HistoryAccessorBase(Configurable):
     """An abstract class for History Accessors """
 
     def get_tail(self, n=10, raw=True, output=False, include_latest=False):
         raise NotImplementedError
 
     def search(self, pattern="*", raw=True, search_raw=True,
                output=False, n=None, unique=False):
         raise NotImplementedError
 
     def get_range(self, session, start=1, stop=None, raw=True,output=False):
         raise NotImplementedError
 
     def get_range_by_str(self, rangestr, raw=True, output=False):
         raise NotImplementedError
 
 
 class HistoryAccessor(HistoryAccessorBase):
     """Access the history database without adding to it.
     
     This is intended for use by standalone history tools. IPython shells use
     HistoryManager, below, which is a subclass of this."""
 
     # String holding the path to the history file
     hist_file = Unicode(config=True,
         help="""Path to file to use for SQLite history database.
         
         By default, IPython will put the history database in the IPython
         profile directory.  If you would rather share one history among
         profiles, you can set this value in each, so that they are consistent.
         
         Due to an issue with fcntl, SQLite is known to misbehave on some NFS
         mounts.  If you see IPython hanging, try setting this to something on a
         local disk, e.g::
         
             ipython --HistoryManager.hist_file=/tmp/ipython_hist.sqlite
         
         """)
     
     enabled = Bool(True, config=True,
         help="""enable the SQLite history
         
         set enabled=False to disable the SQLite history,
         in which case there will be no stored history, no SQLite connection,
         and no background saving thread.  This may be necessary in some
         threaded environments where IPython is embedded.
         """
     )
     
     connection_options = Dict(config=True,
         help="""Options for configuring the SQLite connection
         
         These options are passed as keyword args to sqlite3.connect
         when establishing database conenctions.
         """
     )
 
     # The SQLite database
     db = Any()
     def _db_changed(self, name, old, new):
         """validate the db, since it can be an Instance of two different types"""
         connection_types = (DummyDB,)
         if sqlite3 is not None:
             connection_types = (DummyDB, sqlite3.Connection)
         if not isinstance(new, connection_types):
             msg = "%s.db must be sqlite3 Connection or DummyDB, not %r" % \
                     (self.__class__.__name__, new)
             raise TraitError(msg)
     
     def __init__(self, profile='default', hist_file=u'', **traits):
         """Create a new history accessor.
         
         Parameters
         ----------
         profile : str
           The name of the profile from which to open history.
         hist_file : str
           Path to an SQLite history database stored by IPython. If specified,
           hist_file overrides profile.
         config : :class:`~IPython.config.loader.Config`
           Config object. hist_file can also be set through this.
         """
         # We need a pointer back to the shell for various tasks.
         super(HistoryAccessor, self).__init__(**traits)
         # defer setting hist_file from kwarg until after init,
         # otherwise the default kwarg value would clobber any value
         # set by config
         if hist_file:
             self.hist_file = hist_file
         
         if self.hist_file == u'':
             # No one has set the hist_file, yet.
             self.hist_file = self._get_hist_file_name(profile)
 
         if sqlite3 is None and self.enabled:
             warn("IPython History requires SQLite, your history will not be saved")
             self.enabled = False
         
         self.init_db()
     
     def _get_hist_file_name(self, profile='default'):
         """Find the history file for the given profile name.
         
         This is overridden by the HistoryManager subclass, to use the shell's
         active profile.
         
         Parameters
         ----------
         profile : str
           The name of a profile which has a history file.
         """
         return os.path.join(locate_profile(profile), 'history.sqlite')
     
     @catch_corrupt_db
     def init_db(self):
         """Connect to the database, and create tables if necessary."""
         if not self.enabled:
             self.db = DummyDB()
             return
         
         # use detect_types so that timestamps return datetime objects
         kwargs = dict(detect_types=sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES)
         kwargs.update(self.connection_options)
         self.db = sqlite3.connect(self.hist_file, **kwargs)
         self.db.execute("""CREATE TABLE IF NOT EXISTS sessions (session integer
                         primary key autoincrement, start timestamp,
                         end timestamp, num_cmds integer, remark text)""")
         self.db.execute("""CREATE TABLE IF NOT EXISTS history
                 (session integer, line integer, source text, source_raw text,
                 PRIMARY KEY (session, line))""")
         # Output history is optional, but ensure the table's there so it can be
         # enabled later.
         self.db.execute("""CREATE TABLE IF NOT EXISTS output_history
                         (session integer, line integer, output text,
                         PRIMARY KEY (session, line))""")
         self.db.commit()
 
     def writeout_cache(self):
         """Overridden by HistoryManager to dump the cache before certain
         database lookups."""
         pass
 
     ## -------------------------------
     ## Methods for retrieving history:
     ## -------------------------------
     def _run_sql(self, sql, params, raw=True, output=False):
         """Prepares and runs an SQL query for the history database.
 
         Parameters
         ----------
         sql : str
           Any filtering expressions to go after SELECT ... FROM ...
         params : tuple
           Parameters passed to the SQL query (to replace "?")
         raw, output : bool
           See :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         toget = 'source_raw' if raw else 'source'
         sqlfrom = "history"
         if output:
             sqlfrom = "history LEFT JOIN output_history USING (session, line)"
             toget = "history.%s, output_history.output" % toget
         cur = self.db.execute("SELECT session, line, %s FROM %s " %\
                                 (toget, sqlfrom) + sql, params)
         if output:    # Regroup into 3-tuples, and parse JSON
             return ((ses, lin, (inp, out)) for ses, lin, inp, out in cur)
         return cur
 
     @needs_sqlite
     @catch_corrupt_db
     def get_session_info(self, session):
         """Get info about a session.
 
         Parameters
         ----------
 
         session : int
             Session number to retrieve.
 
         Returns
         -------
         
         session_id : int
            Session ID number
         start : datetime
            Timestamp for the start of the session.
         end : datetime
            Timestamp for the end of the session, or None if IPython crashed.
         num_cmds : int
            Number of commands run, or None if IPython crashed.
         remark : unicode
            A manually set description.
         """
         query = "SELECT * from sessions where session == ?"
         return self.db.execute(query, (session,)).fetchone()
 
     @catch_corrupt_db
     def get_last_session_id(self):
         """Get the last session ID currently in the database.
         
         Within IPython, this should be the same as the value stored in
         :attr:`HistoryManager.session_number`.
         """
         for record in self.get_tail(n=1, include_latest=True):
             return record[0]
 
     @catch_corrupt_db
     def get_tail(self, n=10, raw=True, output=False, include_latest=False):
         """Get the last n lines from the history database.
 
         Parameters
         ----------
         n : int
           The number of lines to get
         raw, output : bool
           See :meth:`get_range`
         include_latest : bool
           If False (default), n+1 lines are fetched, and the latest one
           is discarded. This is intended to be used where the function
           is called by a user command, which it should not return.
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         self.writeout_cache()
         if not include_latest:
             n += 1
         cur = self._run_sql("ORDER BY session DESC, line DESC LIMIT ?",
                                 (n,), raw=raw, output=output)
         if not include_latest:
             return reversed(list(cur)[1:])
         return reversed(list(cur))
 
     @catch_corrupt_db
     def search(self, pattern="*", raw=True, search_raw=True,
                output=False, n=None, unique=False):
         """Search the database using unix glob-style matching (wildcards
         * and ?).
 
         Parameters
         ----------
         pattern : str
           The wildcarded pattern to match when searching
         search_raw : bool
           If True, search the raw input, otherwise, the parsed input
         raw, output : bool
           See :meth:`get_range`
         n : None or int
           If an integer is given, it defines the limit of
           returned entries.
         unique : bool
           When it is true, return only unique entries.
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         tosearch = "source_raw" if search_raw else "source"
         if output:
             tosearch = "history." + tosearch
         self.writeout_cache()
         sqlform = "WHERE %s GLOB ?" % tosearch
         params = (pattern,)
         if unique:
             sqlform += ' GROUP BY {0}'.format(tosearch)
         if n is not None:
             sqlform += " ORDER BY session DESC, line DESC LIMIT ?"
             params += (n,)
         elif unique:
             sqlform += " ORDER BY session, line"
         cur = self._run_sql(sqlform, params, raw=raw, output=output)
         if n is not None:
             return reversed(list(cur))
         return cur
     
     @catch_corrupt_db
     def get_range(self, session, start=1, stop=None, raw=True,output=False):
         """Retrieve input by session.
 
         Parameters
         ----------
         session : int
             Session number to retrieve.
         start : int
             First line to retrieve.
         stop : int
             End of line range (excluded from output itself). If None, retrieve
             to the end of the session.
         raw : bool
             If True, return untranslated input
         output : bool
             If True, attempt to include output. This will be 'real' Python
             objects for the current session, or text reprs from previous
             sessions if db_log_output was enabled at the time. Where no output
             is found, None is used.
 
         Returns
         -------
         entries
           An iterator over the desired lines. Each line is a 3-tuple, either
           (session, line, input) if output is False, or
           (session, line, (input, output)) if output is True.
         """
         if stop:
             lineclause = "line >= ? AND line < ?"
             params = (session, start, stop)
         else:
             lineclause = "line>=?"
             params = (session, start)
 
         return self._run_sql("WHERE session==? AND %s" % lineclause,
                                     params, raw=raw, output=output)
 
     def get_range_by_str(self, rangestr, raw=True, output=False):
         """Get lines of history from a string of ranges, as used by magic
         commands %hist, %save, %macro, etc.
 
         Parameters
         ----------
         rangestr : str
           A string specifying ranges, e.g. "5 ~2/1-4". See
           :func:`magic_history` for full details.
         raw, output : bool
           As :meth:`get_range`
 
         Returns
         -------
         Tuples as :meth:`get_range`
         """
         for sess, s, e in extract_hist_ranges(rangestr):
             for line in self.get_range(sess, s, e, raw=raw, output=output):
                 yield line
 
 
 class HistoryManager(HistoryAccessor):
     """A class to organize all history-related functionality in one place.
     """
     # Public interface
 
     # An instance of the IPython shell we are attached to
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
     # Lists to hold processed and raw history. These start with a blank entry
     # so that we can index them starting from 1
     input_hist_parsed = List([""])
     input_hist_raw = List([""])
     # A list of directories visited during session
     dir_hist = List()
     def _dir_hist_default(self):
         try:
             return [py3compat.getcwd()]
         except OSError:
             return []
 
     # A dict of output history, keyed with ints from the shell's
     # execution count.
     output_hist = Dict()
     # The text/plain repr of outputs.
     output_hist_reprs = Dict()
 
     # The number of the current session in the history database
     session_number = Integer()
     
     db_log_output = Bool(False, config=True,
         help="Should the history database include output? (default: no)"
     )
     db_cache_size = Integer(0, config=True,
         help="Write to database every x commands (higher values save disk access & power).\n"
         "Values of 1 or less effectively disable caching."
     )
     # The input and output caches
     db_input_cache = List()
     db_output_cache = List()
     
     # History saving in separate thread
     save_thread = Instance('IPython.core.history.HistorySavingThread')
     try:               # Event is a function returning an instance of _Event...
         save_flag = Instance(threading._Event)
     except AttributeError:         # ...until Python 3.3, when it's a class.
         save_flag = Instance(threading.Event)
     
     # Private interface
     # Variables used to store the three last inputs from the user.  On each new
     # history update, we populate the user's namespace with these, shifted as
     # necessary.
     _i00 = Unicode(u'')
     _i = Unicode(u'')
     _ii = Unicode(u'')
     _iii = Unicode(u'')
 
     # A regex matching all forms of the exit command, so that we don't store
     # them in the history (it's annoying to rewind the first entry and land on
     # an exit call).
     _exit_re = re.compile(r"(exit|quit)(\s*\(.*\))?$")
 
     def __init__(self, shell=None, config=None, **traits):
         """Create a new history manager associated with a shell instance.
         """
         # We need a pointer back to the shell for various tasks.
         super(HistoryManager, self).__init__(shell=shell, config=config,
             **traits)
         self.save_flag = threading.Event()
         self.db_input_cache_lock = threading.Lock()
         self.db_output_cache_lock = threading.Lock()
         if self.enabled and self.hist_file != ':memory:':
             self.save_thread = HistorySavingThread(self)
             self.save_thread.start()
 
         self.new_session()
 
     def _get_hist_file_name(self, profile=None):
         """Get default history file name based on the Shell's profile.
         
         The profile parameter is ignored, but must exist for compatibility with
         the parent class."""
         profile_dir = self.shell.profile_dir.location
         return os.path.join(profile_dir, 'history.sqlite')
     
     @needs_sqlite
     def new_session(self, conn=None):
         """Get a new session number."""
         if conn is None:
             conn = self.db
         
         with conn:
             cur = conn.execute("""INSERT INTO sessions VALUES (NULL, ?, NULL,
                             NULL, "") """, (datetime.datetime.now(),))
             self.session_number = cur.lastrowid
             
     def end_session(self):
         """Close the database session, filling in the end time and line count."""
         self.writeout_cache()
         with self.db:
             self.db.execute("""UPDATE sessions SET end=?, num_cmds=? WHERE
                             session==?""", (datetime.datetime.now(),
                             len(self.input_hist_parsed)-1, self.session_number))
         self.session_number = 0
                             
     def name_session(self, name):
         """Give the current session a name in the history database."""
         with self.db:
             self.db.execute("UPDATE sessions SET remark=? WHERE session==?",
                             (name, self.session_number))
                             
     def reset(self, new_session=True):
         """Clear the session history, releasing all object references, and
         optionally open a new session."""
         self.output_hist.clear()
         # The directory history can't be completely empty
         self.dir_hist[:] = [py3compat.getcwd()]
         
         if new_session:
             if self.session_number:
                 self.end_session()
             self.input_hist_parsed[:] = [""]
             self.input_hist_raw[:] = [""]
             self.new_session()
 
     # ------------------------------
     # Methods for retrieving history
     # ------------------------------
     def get_session_info(self, session=0):
         """Get info about a session.
 
         Parameters
         ----------
 
         session : int
             Session number to retrieve. The current session is 0, and negative
             numbers count back from current session, so -1 is the previous session.
 
         Returns
         -------
         
         session_id : int
            Session ID number
         start : datetime
            Timestamp for the start of the session.
         end : datetime
            Timestamp for the end of the session, or None if IPython crashed.
         num_cmds : int
            Number of commands run, or None if IPython crashed.
         remark : unicode
            A manually set description.
         """
         if session <= 0:
             session += self.session_number
 
         return super(HistoryManager, self).get_session_info(session=session)
 
     def _get_range_session(self, start=1, stop=None, raw=True, output=False):
         """Get input and output history from the current session. Called by
         get_range, and takes similar parameters."""
         input_hist = self.input_hist_raw if raw else self.input_hist_parsed
             
         n = len(input_hist)
         if start < 0:
             start += n
         if not stop or (stop > n):
             stop = n
         elif stop < 0:
             stop += n
         
         for i in range(start, stop):
             if output:
                 line = (input_hist[i], self.output_hist_reprs.get(i))
             else:
                 line = input_hist[i]
             yield (0, i, line)
     
     def get_range(self, session=0, start=1, stop=None, raw=True,output=False):
         """Retrieve input by session.
         
         Parameters
         ----------
         session : int
             Session number to retrieve. The current session is 0, and negative
             numbers count back from current session, so -1 is previous session.
         start : int
             First line to retrieve.
         stop : int
             End of line range (excluded from output itself). If None, retrieve
             to the end of the session.
         raw : bool
             If True, return untranslated input
         output : bool
             If True, attempt to include output. This will be 'real' Python
             objects for the current session, or text reprs from previous
             sessions if db_log_output was enabled at the time. Where no output
             is found, None is used.
             
         Returns
         -------
         entries
           An iterator over the desired lines. Each line is a 3-tuple, either
           (session, line, input) if output is False, or
           (session, line, (input, output)) if output is True.
         """
         if session <= 0:
             session += self.session_number
         if session==self.session_number:          # Current session
             return self._get_range_session(start, stop, raw, output)
         return super(HistoryManager, self).get_range(session, start, stop, raw,
                                                      output)
 
     ## ----------------------------
     ## Methods for storing history:
     ## ----------------------------
     def store_inputs(self, line_num, source, source_raw=None):
         """Store source and raw input in history and create input cache
         variables ``_i*``.
 
         Parameters
         ----------
         line_num : int
           The prompt number of this input.
 
         source : str
           Python input.
 
         source_raw : str, optional
           If given, this is the raw input without any IPython transformations
           applied to it.  If not given, ``source`` is used.
         """
         if source_raw is None:
             source_raw = source
         source = source.rstrip('\n')
         source_raw = source_raw.rstrip('\n')
 
         # do not store exit/quit commands
         if self._exit_re.match(source_raw.strip()):
             return
 
         self.input_hist_parsed.append(source)
         self.input_hist_raw.append(source_raw)
 
         with self.db_input_cache_lock:
             self.db_input_cache.append((line_num, source, source_raw))
             # Trigger to flush cache and write to DB.
             if len(self.db_input_cache) >= self.db_cache_size:
                 self.save_flag.set()
 
         # update the auto _i variables
         self._iii = self._ii
         self._ii = self._i
         self._i = self._i00
         self._i00 = source_raw
 
         # hackish access to user namespace to create _i1,_i2... dynamically
         new_i = '_i%s' % line_num
         to_main = {'_i': self._i,
                    '_ii': self._ii,
                    '_iii': self._iii,
                    new_i : self._i00 }
         
         if self.shell is not None:
             self.shell.push(to_main, interactive=False)
 
     def store_output(self, line_num):
         """If database output logging is enabled, this saves all the
         outputs from the indicated prompt number to the database. It's
         called by run_cell after code has been executed.
 
         Parameters
         ----------
         line_num : int
           The line number from which to save outputs
         """
         if (not self.db_log_output) or (line_num not in self.output_hist_reprs):
             return
         output = self.output_hist_reprs[line_num]
 
         with self.db_output_cache_lock:
             self.db_output_cache.append((line_num, output))
         if self.db_cache_size <= 1:
             self.save_flag.set()
 
     def _writeout_input_cache(self, conn):
         with conn:
             for line in self.db_input_cache:
                 conn.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
                                 (self.session_number,)+line)
 
     def _writeout_output_cache(self, conn):
         with conn:
             for line in self.db_output_cache:
                 conn.execute("INSERT INTO output_history VALUES (?, ?, ?)",
                                 (self.session_number,)+line)
 
     @needs_sqlite
     def writeout_cache(self, conn=None):
         """Write any entries in the cache to the database."""
         if conn is None:
             conn = self.db
 
         with self.db_input_cache_lock:
             try:
                 self._writeout_input_cache(conn)
             except sqlite3.IntegrityError:
                 self.new_session(conn)
                 print("ERROR! Session/line number was not unique in",
                       "database. History logging moved to new session",
                                                 self.session_number)
                 try:
                     # Try writing to the new session. If this fails, don't
                     # recurse
                     self._writeout_input_cache(conn)
                 except sqlite3.IntegrityError:
                     pass
             finally:
                 self.db_input_cache = []
 
         with self.db_output_cache_lock:
             try:
                 self._writeout_output_cache(conn)
             except sqlite3.IntegrityError:
                 print("!! Session/line number for output was not unique",
                       "in database. Output will not be stored.")
             finally:
                 self.db_output_cache = []
 
 
 class HistorySavingThread(threading.Thread):
     """This thread takes care of writing history to the database, so that
     the UI isn't held up while that happens.
 
     It waits for the HistoryManager's save_flag to be set, then writes out
     the history cache. The main thread is responsible for setting the flag when
     the cache size reaches a defined threshold."""
     daemon = True
     stop_now = False
     enabled = True
     def __init__(self, history_manager):
         super(HistorySavingThread, self).__init__(name="IPythonHistorySavingThread")
         self.history_manager = history_manager
         self.enabled = history_manager.enabled
         atexit.register(self.stop)
 
     @needs_sqlite
     def run(self):
         # We need a separate db connection per thread:
         try:
             self.db = sqlite3.connect(self.history_manager.hist_file,
                             **self.history_manager.connection_options
             )
             while True:
                 self.history_manager.save_flag.wait()
                 if self.stop_now:
                     self.db.close()
                     return
                 self.history_manager.save_flag.clear()
                 self.history_manager.writeout_cache(self.db)
         except Exception as e:
             print(("The history saving thread hit an unexpected error (%s)."
                    "History will not be written to the database.") % repr(e))
 
     def stop(self):
         """This can be called from the main thread to safely stop this thread.
 
         Note that it does not attempt to write out remaining history before
         exiting. That should be done by calling the HistoryManager's
         end_session method."""
         self.stop_now = True
         self.history_manager.save_flag.set()
         self.join()
 
 
 # To match, e.g. ~5/8-~2/3
 range_re = re.compile(r"""
 ((?P<startsess>~?\d+)/)?
 (?P<start>\d+)?
 ((?P<sep>[\-:])
  ((?P<endsess>~?\d+)/)?
  (?P<end>\d+))?
 $""", re.VERBOSE)
 
 
 def extract_hist_ranges(ranges_str):
     """Turn a string of history ranges into 3-tuples of (session, start, stop).
 
     Examples
     --------
     >>> list(extract_hist_ranges("~8/5-~7/4 2"))
     [(-8, 5, None), (-7, 1, 5), (0, 2, 3)]
     """
     for range_str in ranges_str.split():
         rmatch = range_re.match(range_str)
         if not rmatch:
             continue
         start = rmatch.group("start")
         if start:
             start = int(start)
             end = rmatch.group("end")
             # If no end specified, get (a, a + 1)
             end = int(end) if end else start + 1
         else:  # start not specified
             if not rmatch.group('startsess'):  # no startsess
                 continue
             start = 1
             end = None  # provide the entire session hist
 
         if rmatch.group("sep") == "-":       # 1-3 == 1:4 --> [1, 2, 3]
             end += 1
         startsess = rmatch.group("startsess") or "0"
         endsess = rmatch.group("endsess") or startsess
         startsess = int(startsess.replace("~","-"))
         endsess = int(endsess.replace("~","-"))
         assert endsess >= startsess, "start session must be earlier than end session"
 
         if endsess == startsess:
             yield (startsess, start, end)
             continue
         # Multiple sessions in one range:
         yield (startsess, start, None)
         for sess in range(startsess+1, endsess):
             yield (sess, 1, None)
         yield (endsess, 1, end)
 
 
 def _format_lineno(session, line):
     """Helper function to format line numbers properly."""
     if session == 0:
         return str(line)
     return "%s#%s" % (session, line)
 
 

IPython/core/magic.py

@@ -1,702 +1,702 b''
 # encoding: utf-8
 """Magic functions for InteractiveShell.
 """
 from __future__ import print_function
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2001 Janko Hauser <[e-mail]> and
 #  Copyright (C) 2001 Fernando Perez <[e-mail]>
 #  Copyright (C) 2008 The IPython Development Team
 
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 # Stdlib
 import os
 import re
 import sys
 import types
 from getopt import getopt, GetoptError
 
 # Our own
 from IPython.config.configurable import Configurable
 from IPython.core import oinspect
 from IPython.core.error import UsageError
 from IPython.core.inputsplitter import ESC_MAGIC, ESC_MAGIC2
-from IPython.external.decorator import decorator
+from decorator import decorator
 from IPython.utils.ipstruct import Struct
 from IPython.utils.process import arg_split
 from IPython.utils.py3compat import string_types, iteritems
 from IPython.utils.text import dedent
 from IPython.utils.traitlets import Bool, Dict, Instance, MetaHasTraits
 from IPython.utils.warn import error
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 # A dict we'll use for each class that has magics, used as temporary storage to
 # pass information between the @line/cell_magic method decorators and the
 # @magics_class class decorator, because the method decorators have no
 # access to the class when they run.  See for more details:
 # http://stackoverflow.com/questions/2366713/can-a-python-decorator-of-an-instance-method-access-the-class
 
 magics = dict(line={}, cell={})
 
 magic_kinds = ('line', 'cell')
 magic_spec = ('line', 'cell', 'line_cell')
 magic_escapes = dict(line=ESC_MAGIC, cell=ESC_MAGIC2)
 
 #-----------------------------------------------------------------------------
 # Utility classes and functions
 #-----------------------------------------------------------------------------
 
 class Bunch: pass
 
 
 def on_off(tag):
     """Return an ON/OFF string for a 1/0 input. Simple utility function."""
     return ['OFF','ON'][tag]
 
 
 def compress_dhist(dh):
     """Compress a directory history into a new one with at most 20 entries.
 
     Return a new list made from the first and last 10 elements of dhist after
     removal of duplicates.
     """
     head, tail = dh[:-10], dh[-10:]
 
     newhead = []
     done = set()
     for h in head:
         if h in done:
             continue
         newhead.append(h)
         done.add(h)
 
     return newhead + tail
 
 
 def needs_local_scope(func):
     """Decorator to mark magic functions which need to local scope to run."""
     func.needs_local_scope = True
     return func
 
 #-----------------------------------------------------------------------------
 # Class and method decorators for registering magics
 #-----------------------------------------------------------------------------
 
 def magics_class(cls):
     """Class decorator for all subclasses of the main Magics class.
 
     Any class that subclasses Magics *must* also apply this decorator, to
     ensure that all the methods that have been decorated as line/cell magics
     get correctly registered in the class instance.  This is necessary because
     when method decorators run, the class does not exist yet, so they
     temporarily store their information into a module global.  Application of
     this class decorator copies that global data to the class instance and
     clears the global.
 
     Obviously, this mechanism is not thread-safe, which means that the
     *creation* of subclasses of Magic should only be done in a single-thread
     context.  Instantiation of the classes has no restrictions.  Given that
     these classes are typically created at IPython startup time and before user
     application code becomes active, in practice this should not pose any
     problems.
     """
     cls.registered = True
     cls.magics = dict(line = magics['line'],
                       cell = magics['cell'])
     magics['line'] = {}
     magics['cell'] = {}
     return cls
 
 
 def record_magic(dct, magic_kind, magic_name, func):
     """Utility function to store a function as a magic of a specific kind.
 
     Parameters
     ----------
     dct : dict
       A dictionary with 'line' and 'cell' subdicts.
 
     magic_kind : str
       Kind of magic to be stored.
 
     magic_name : str
       Key to store the magic as.
 
     func : function
       Callable object to store.
     """
     if magic_kind == 'line_cell':
         dct['line'][magic_name] = dct['cell'][magic_name] = func
     else:
         dct[magic_kind][magic_name] = func
 
 
 def validate_type(magic_kind):
     """Ensure that the given magic_kind is valid.
 
     Check that the given magic_kind is one of the accepted spec types (stored
     in the global `magic_spec`), raise ValueError otherwise.
     """
     if magic_kind not in magic_spec:
         raise ValueError('magic_kind must be one of %s, %s given' %
                          magic_kinds, magic_kind)
 
 
 # The docstrings for the decorator below will be fairly similar for the two
 # types (method and function), so we generate them here once and reuse the
 # templates below.
 _docstring_template = \
 """Decorate the given {0} as {1} magic.
 
 The decorator can be used with or without arguments, as follows.
 
 i) without arguments: it will create a {1} magic named as the {0} being
 decorated::
 
     @deco
     def foo(...)
 
 will create a {1} magic named `foo`.
 
 ii) with one string argument: which will be used as the actual name of the
 resulting magic::
 
     @deco('bar')
     def foo(...)
 
 will create a {1} magic named `bar`.
 """
 
 # These two are decorator factories.  While they are conceptually very similar,
 # there are enough differences in the details that it's simpler to have them
 # written as completely standalone functions rather than trying to share code
 # and make a single one with convoluted logic.
 
 def _method_magic_marker(magic_kind):
     """Decorator factory for methods in Magics subclasses.
     """
 
     validate_type(magic_kind)
 
     # This is a closure to capture the magic_kind.  We could also use a class,
     # but it's overkill for just that one bit of state.
     def magic_deco(arg):
         call = lambda f, *a, **k: f(*a, **k)
 
         if callable(arg):
             # "Naked" decorator call (just @foo, no args)
             func = arg
             name = func.__name__
             retval = decorator(call, func)
             record_magic(magics, magic_kind, name, name)
         elif isinstance(arg, string_types):
             # Decorator called with arguments (@foo('bar'))
             name = arg
             def mark(func, *a, **kw):
                 record_magic(magics, magic_kind, name, func.__name__)
                 return decorator(call, func)
             retval = mark
         else:
             raise TypeError("Decorator can only be called with "
                             "string or function")
         return retval
 
     # Ensure the resulting decorator has a usable docstring
     magic_deco.__doc__ = _docstring_template.format('method', magic_kind)
     return magic_deco
 
 
 def _function_magic_marker(magic_kind):
     """Decorator factory for standalone functions.
     """
     validate_type(magic_kind)
     
     # This is a closure to capture the magic_kind.  We could also use a class,
     # but it's overkill for just that one bit of state.
     def magic_deco(arg):
         call = lambda f, *a, **k: f(*a, **k)
 
         # Find get_ipython() in the caller's namespace
         caller = sys._getframe(1)
         for ns in ['f_locals', 'f_globals', 'f_builtins']:
             get_ipython = getattr(caller, ns).get('get_ipython')
             if get_ipython is not None:
                 break
         else:
             raise NameError('Decorator can only run in context where '
                             '`get_ipython` exists')
 
         ip = get_ipython()
 
         if callable(arg):
             # "Naked" decorator call (just @foo, no args)
             func = arg
             name = func.__name__
             ip.register_magic_function(func, magic_kind, name)
             retval = decorator(call, func)
         elif isinstance(arg, string_types):
             # Decorator called with arguments (@foo('bar'))
             name = arg
             def mark(func, *a, **kw):
                 ip.register_magic_function(func, magic_kind, name)
                 return decorator(call, func)
             retval = mark
         else:
             raise TypeError("Decorator can only be called with "
                              "string or function")
         return retval
 
     # Ensure the resulting decorator has a usable docstring
     ds = _docstring_template.format('function', magic_kind)
 
     ds += dedent("""
     Note: this decorator can only be used in a context where IPython is already
     active, so that the `get_ipython()` call succeeds.  You can therefore use
     it in your startup files loaded after IPython initializes, but *not* in the
     IPython configuration file itself, which is executed before IPython is
     fully up and running.  Any file located in the `startup` subdirectory of
     your configuration profile will be OK in this sense.
     """)
     
     magic_deco.__doc__ = ds
     return magic_deco
 
 
 # Create the actual decorators for public use
 
 # These three are used to decorate methods in class definitions
 line_magic = _method_magic_marker('line')
 cell_magic = _method_magic_marker('cell')
 line_cell_magic = _method_magic_marker('line_cell')
 
 # These three decorate standalone functions and perform the decoration
 # immediately.  They can only run where get_ipython() works
 register_line_magic = _function_magic_marker('line')
 register_cell_magic = _function_magic_marker('cell')
 register_line_cell_magic = _function_magic_marker('line_cell')
 
 #-----------------------------------------------------------------------------
 # Core Magic classes
 #-----------------------------------------------------------------------------
 
 class MagicsManager(Configurable):
     """Object that handles all magic-related functionality for IPython.
     """
     # Non-configurable class attributes
 
     # A two-level dict, first keyed by magic type, then by magic function, and
     # holding the actual callable object as value.  This is the dict used for
     # magic function dispatch
     magics = Dict
 
     # A registry of the original objects that we've been given holding magics.
     registry = Dict
 
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
 
     auto_magic = Bool(True, config=True, help=
         "Automatically call line magics without requiring explicit % prefix")
 
     def _auto_magic_changed(self, name, value):
         self.shell.automagic = value
     
     _auto_status = [
         'Automagic is OFF, % prefix IS needed for line magics.',
         'Automagic is ON, % prefix IS NOT needed for line magics.']
 
     user_magics = Instance('IPython.core.magics.UserMagics')
 
     def __init__(self, shell=None, config=None, user_magics=None, **traits):
 
         super(MagicsManager, self).__init__(shell=shell, config=config,
                                            user_magics=user_magics, **traits)
         self.magics = dict(line={}, cell={})
         # Let's add the user_magics to the registry for uniformity, so *all*
         # registered magic containers can be found there.
         self.registry[user_magics.__class__.__name__] = user_magics
 
     def auto_status(self):
         """Return descriptive string with automagic status."""
         return self._auto_status[self.auto_magic]
     
     def lsmagic(self):
         """Return a dict of currently available magic functions.
 
         The return dict has the keys 'line' and 'cell', corresponding to the
         two types of magics we support.  Each value is a list of names.
         """
         return self.magics
 
     def lsmagic_docs(self, brief=False, missing=''):
         """Return dict of documentation of magic functions.
 
         The return dict has the keys 'line' and 'cell', corresponding to the
         two types of magics we support. Each value is a dict keyed by magic
         name whose value is the function docstring. If a docstring is
         unavailable, the value of `missing` is used instead.
 
         If brief is True, only the first line of each docstring will be returned.
         """
         docs = {}
         for m_type in self.magics:
             m_docs = {}
             for m_name, m_func in iteritems(self.magics[m_type]):
                 if m_func.__doc__:
                     if brief:
                         m_docs[m_name] = m_func.__doc__.split('\n', 1)[0]
                     else:
                         m_docs[m_name] = m_func.__doc__.rstrip()
                 else:
                     m_docs[m_name] = missing
             docs[m_type] = m_docs
         return docs
 
     def register(self, *magic_objects):
         """Register one or more instances of Magics.
 
         Take one or more classes or instances of classes that subclass the main 
         `core.Magic` class, and register them with IPython to use the magic
         functions they provide.  The registration process will then ensure that
         any methods that have decorated to provide line and/or cell magics will
         be recognized with the `%x`/`%%x` syntax as a line/cell magic
         respectively.
 
         If classes are given, they will be instantiated with the default
         constructor.  If your classes need a custom constructor, you should
         instanitate them first and pass the instance.
 
         The provided arguments can be an arbitrary mix of classes and instances.
 
         Parameters
         ----------
         magic_objects : one or more classes or instances
         """
         # Start by validating them to ensure they have all had their magic
         # methods registered at the instance level
         for m in magic_objects:
             if not m.registered:
                 raise ValueError("Class of magics %r was constructed without "
                                  "the @register_magics class decorator")
             if type(m) in (type, MetaHasTraits):
                 # If we're given an uninstantiated class
                 m = m(shell=self.shell)
 
             # Now that we have an instance, we can register it and update the
             # table of callables
             self.registry[m.__class__.__name__] = m
             for mtype in magic_kinds:
                 self.magics[mtype].update(m.magics[mtype])
 
     def register_function(self, func, magic_kind='line', magic_name=None):
         """Expose a standalone function as magic function for IPython.
 
         This will create an IPython magic (line, cell or both) from a
         standalone function.  The functions should have the following
         signatures: 
 
         * For line magics: `def f(line)`
         * For cell magics: `def f(line, cell)`
         * For a function that does both: `def f(line, cell=None)`
 
         In the latter case, the function will be called with `cell==None` when
         invoked as `%f`, and with cell as a string when invoked as `%%f`.
 
         Parameters
         ----------
         func : callable
           Function to be registered as a magic.
 
         magic_kind : str
           Kind of magic, one of 'line', 'cell' or 'line_cell'
 
         magic_name : optional str
           If given, the name the magic will have in the IPython namespace.  By
           default, the name of the function itself is used.
         """
 
         # Create the new method in the user_magics and register it in the
         # global table
         validate_type(magic_kind)
         magic_name = func.__name__ if magic_name is None else magic_name
         setattr(self.user_magics, magic_name, func)
         record_magic(self.magics, magic_kind, magic_name, func)
 
     def define_magic(self, name, func):
         """[Deprecated] Expose own function as magic function for IPython.
 
         Example::
 
             def foo_impl(self, parameter_s=''):
                 'My very own magic!. (Use docstrings, IPython reads them).'
                 print 'Magic function. Passed parameter is between < >:'
                 print '<%s>' % parameter_s
                 print 'The self object is:', self
 
             ip.define_magic('foo',foo_impl)
         """
         meth = types.MethodType(func, self.user_magics)
         setattr(self.user_magics, name, meth)
         record_magic(self.magics, 'line', name, meth)
 
     def register_alias(self, alias_name, magic_name, magic_kind='line'):
         """Register an alias to a magic function.
 
         The alias is an instance of :class:`MagicAlias`, which holds the
         name and kind of the magic it should call. Binding is done at
         call time, so if the underlying magic function is changed the alias
         will call the new function.
 
         Parameters
         ----------
         alias_name : str
           The name of the magic to be registered.
 
         magic_name : str
           The name of an existing magic.
 
         magic_kind : str
           Kind of magic, one of 'line' or 'cell'
         """
 
         # `validate_type` is too permissive, as it allows 'line_cell'
         # which we do not handle.
         if magic_kind not in magic_kinds:
             raise ValueError('magic_kind must be one of %s, %s given' %
                              magic_kinds, magic_kind)
 
         alias = MagicAlias(self.shell, magic_name, magic_kind)
         setattr(self.user_magics, alias_name, alias)
         record_magic(self.magics, magic_kind, alias_name, alias)
 
 # Key base class that provides the central functionality for magics.
 
 
 class Magics(Configurable):
     """Base class for implementing magic functions.
 
     Shell functions which can be reached as %function_name. All magic
     functions should accept a string, which they can parse for their own
     needs. This can make some functions easier to type, eg `%cd ../`
     vs. `%cd("../")`
 
     Classes providing magic functions need to subclass this class, and they
     MUST:
 
     - Use the method decorators `@line_magic` and `@cell_magic` to decorate
       individual methods as magic functions, AND
 
     - Use the class decorator `@magics_class` to ensure that the magic
       methods are properly registered at the instance level upon instance
       initialization.
 
     See :mod:`magic_functions` for examples of actual implementation classes.
     """
     # Dict holding all command-line options for each magic.
     options_table = None
     # Dict for the mapping of magic names to methods, set by class decorator
     magics = None
     # Flag to check that the class decorator was properly applied
     registered = False
     # Instance of IPython shell
     shell = None
 
     def __init__(self, shell=None, **kwargs):
         if not(self.__class__.registered):
             raise ValueError('Magics subclass without registration - '
                              'did you forget to apply @magics_class?')
         if shell is not None:
             if hasattr(shell, 'configurables'):
                 shell.configurables.append(self)
             if hasattr(shell, 'config'):
                 kwargs.setdefault('parent', shell)
             kwargs['shell'] = shell
 
         self.shell = shell
         self.options_table = {}
         # The method decorators are run when the instance doesn't exist yet, so
         # they can only record the names of the methods they are supposed to
         # grab.  Only now, that the instance exists, can we create the proper
         # mapping to bound methods.  So we read the info off the original names
         # table and replace each method name by the actual bound method.
         # But we mustn't clobber the *class* mapping, in case of multiple instances.
         class_magics = self.magics
         self.magics = {}
         for mtype in magic_kinds:
             tab = self.magics[mtype] = {}
             cls_tab = class_magics[mtype]
             for magic_name, meth_name in iteritems(cls_tab):
                 if isinstance(meth_name, string_types):
                     # it's a method name, grab it
                     tab[magic_name] = getattr(self, meth_name)
                 else:
                     # it's the real thing
                     tab[magic_name] = meth_name
         # Configurable **needs** to be initiated at the end or the config
         # magics get screwed up.
         super(Magics, self).__init__(**kwargs)
 
     def arg_err(self,func):
         """Print docstring if incorrect arguments were passed"""
         print('Error in arguments:')
         print(oinspect.getdoc(func))
 
     def format_latex(self, strng):
         """Format a string for latex inclusion."""
 
         # Characters that need to be escaped for latex:
         escape_re = re.compile(r'(%|_|\$|#|&)',re.MULTILINE)
         # Magic command names as headers:
         cmd_name_re = re.compile(r'^(%s.*?):' % ESC_MAGIC,
                                  re.MULTILINE)
         # Magic commands
         cmd_re = re.compile(r'(?P<cmd>%s.+?\b)(?!\}\}:)' % ESC_MAGIC,
                             re.MULTILINE)
         # Paragraph continue
         par_re = re.compile(r'\\$',re.MULTILINE)
 
         # The "\n" symbol
         newline_re = re.compile(r'\\n')
 
         # Now build the string for output:
         #strng = cmd_name_re.sub(r'\n\\texttt{\\textsl{\\large \1}}:',strng)
         strng = cmd_name_re.sub(r'\n\\bigskip\n\\texttt{\\textbf{ \1}}:',
                                 strng)
         strng = cmd_re.sub(r'\\texttt{\g<cmd>}',strng)
         strng = par_re.sub(r'\\\\',strng)
         strng = escape_re.sub(r'\\\1',strng)
         strng = newline_re.sub(r'\\textbackslash{}n',strng)
         return strng
 
     def parse_options(self, arg_str, opt_str, *long_opts, **kw):
         """Parse options passed to an argument string.
 
         The interface is similar to that of :func:`getopt.getopt`, but it
         returns a :class:`~IPython.utils.struct.Struct` with the options as keys
         and the stripped argument string still as a string.
 
         arg_str is quoted as a true sys.argv vector by using shlex.split.
         This allows us to easily expand variables, glob files, quote
         arguments, etc.
 
         Parameters
         ----------
 
         arg_str : str
           The arguments to parse.
 
         opt_str : str
           The options specification.
 
         mode : str, default 'string'
           If given as 'list', the argument string is returned as a list (split
           on whitespace) instead of a string.
 
         list_all : bool, default False
           Put all option values in lists. Normally only options
           appearing more than once are put in a list.
 
         posix : bool, default True
           Whether to split the input line in POSIX mode or not, as per the
           conventions outlined in the :mod:`shlex` module from the standard
           library.
         """
 
         # inject default options at the beginning of the input line
         caller = sys._getframe(1).f_code.co_name
         arg_str = '%s %s' % (self.options_table.get(caller,''),arg_str)
 
         mode = kw.get('mode','string')
         if mode not in ['string','list']:
             raise ValueError('incorrect mode given: %s' % mode)
         # Get options
         list_all = kw.get('list_all',0)
         posix = kw.get('posix', os.name == 'posix')
         strict = kw.get('strict', True)
 
         # Check if we have more than one argument to warrant extra processing:
         odict = {}  # Dictionary with options
         args = arg_str.split()
         if len(args) >= 1:
             # If the list of inputs only has 0 or 1 thing in it, there's no
             # need to look for options
             argv = arg_split(arg_str, posix, strict)
             # Do regular option processing
             try:
                 opts,args = getopt(argv, opt_str, long_opts)
             except GetoptError as e:
                 raise UsageError('%s ( allowed: "%s" %s)' % (e.msg,opt_str,
                                         " ".join(long_opts)))
             for o,a in opts:
                 if o.startswith('--'):
                     o = o[2:]
                 else:
                     o = o[1:]
                 try:
                     odict[o].append(a)
                 except AttributeError:
                     odict[o] = [odict[o],a]
                 except KeyError:
                     if list_all:
                         odict[o] = [a]
                     else:
                         odict[o] = a
 
         # Prepare opts,args for return
         opts = Struct(odict)
         if mode == 'string':
             args = ' '.join(args)
 
         return opts,args
 
     def default_option(self, fn, optstr):
         """Make an entry in the options_table for fn, with value optstr"""
 
         if fn not in self.lsmagic():
             error("%s is not a magic function" % fn)
         self.options_table[fn] = optstr
 
 
 class MagicAlias(object):
     """An alias to another magic function.
 
     An alias is determined by its magic name and magic kind. Lookup
     is done at call time, so if the underlying magic changes the alias
     will call the new function.
 
     Use the :meth:`MagicsManager.register_alias` method or the
     `%alias_magic` magic function to create and register a new alias.
     """
     def __init__(self, shell, magic_name, magic_kind):
         self.shell = shell
         self.magic_name = magic_name
         self.magic_kind = magic_kind
 
         self.pretty_target = '%s%s' % (magic_escapes[self.magic_kind], self.magic_name)
         self.__doc__ = "Alias for `%s`." % self.pretty_target
 
         self._in_call = False
 
     def __call__(self, *args, **kwargs):
         """Call the magic alias."""
         fn = self.shell.find_magic(self.magic_name, self.magic_kind)
         if fn is None:
             raise UsageError("Magic `%s` not found." % self.pretty_target)
 
         # Protect against infinite recursion.
         if self._in_call:
             raise UsageError("Infinite recursion detected; "
                              "magic aliases cannot call themselves.")
         self._in_call = True
         try:
             return fn(*args, **kwargs)
         finally:
             self._in_call = False

IPython/core/tests/test_oinspect.py

@@ -1,382 +1,382 b''
 """Tests for the object inspection functionality.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011 The IPython Development Team.
 #
 #  Distributed under the terms of the BSD License.
 #
 #  The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib imports
 import os
 import re
 
 # Third-party imports
 import nose.tools as nt
 
 # Our own imports
 from .. import oinspect
 from IPython.core.magic import (Magics, magics_class, line_magic,
                                 cell_magic, line_cell_magic,
                                 register_line_magic, register_cell_magic,
                                 register_line_cell_magic)
-from IPython.external.decorator import decorator
+from decorator import decorator
 from IPython.testing.decorators import skipif
 from IPython.testing.tools import AssertPrints
 from IPython.utils.path import compress_user
 from IPython.utils import py3compat
 
 
 #-----------------------------------------------------------------------------
 # Globals and constants
 #-----------------------------------------------------------------------------
 
 inspector = oinspect.Inspector()
 ip = get_ipython()
 
 #-----------------------------------------------------------------------------
 # Local utilities
 #-----------------------------------------------------------------------------
 
 # WARNING: since this test checks the line number where a function is
 # defined, if any code is inserted above, the following line will need to be
 # updated.  Do NOT insert any whitespace between the next line and the function
 # definition below.
 THIS_LINE_NUMBER = 51  # Put here the actual number of this line
 def test_find_source_lines():
     nt.assert_equal(oinspect.find_source_lines(test_find_source_lines), 
                     THIS_LINE_NUMBER+1)
 
 
 # A couple of utilities to ensure these tests work the same from a source or a
 # binary install
 def pyfile(fname):
     return os.path.normcase(re.sub('.py[co]$', '.py', fname))
 
 
 def match_pyfiles(f1, f2):
     nt.assert_equal(pyfile(f1), pyfile(f2))
 
 
 def test_find_file():
     match_pyfiles(oinspect.find_file(test_find_file), os.path.abspath(__file__))
 
 
 def test_find_file_decorated1():
 
     @decorator
     def noop1(f):
         def wrapper():
             return f(*a, **kw)
         return wrapper
 
     @noop1
     def f(x):
         "My docstring"
     
     match_pyfiles(oinspect.find_file(f), os.path.abspath(__file__))
     nt.assert_equal(f.__doc__, "My docstring")
 
 
 def test_find_file_decorated2():
 
     @decorator
     def noop2(f, *a, **kw):
         return f(*a, **kw)
 
     @noop2
     def f(x):
         "My docstring 2"
     
     match_pyfiles(oinspect.find_file(f), os.path.abspath(__file__))
     nt.assert_equal(f.__doc__, "My docstring 2")
     
 
 def test_find_file_magic():
     run = ip.find_line_magic('run')
     nt.assert_not_equal(oinspect.find_file(run), None)
 
 
 # A few generic objects we can then inspect in the tests below
 
 class Call(object):
     """This is the class docstring."""
 
     def __init__(self, x, y=1):
         """This is the constructor docstring."""
 
     def __call__(self, *a, **kw):
         """This is the call docstring."""
 
     def method(self, x, z=2):
         """Some method's docstring"""
 
 class SimpleClass(object):
     def method(self, x, z=2):
         """Some method's docstring"""
 
 
 class OldStyle:
     """An old-style class for testing."""
     pass
 
 
 def f(x, y=2, *a, **kw):
     """A simple function."""
 
 
 def g(y, z=3, *a, **kw):
     pass  # no docstring
 
 
 @register_line_magic
 def lmagic(line):
     "A line magic"
 
 
 @register_cell_magic
 def cmagic(line, cell):
     "A cell magic"
 
 
 @register_line_cell_magic
 def lcmagic(line, cell=None):
     "A line/cell magic"
 
 
 @magics_class
 class SimpleMagics(Magics):
     @line_magic
     def Clmagic(self, cline):
         "A class-based line magic"
         
     @cell_magic
     def Ccmagic(self, cline, ccell):
         "A class-based cell magic"
         
     @line_cell_magic
     def Clcmagic(self, cline, ccell=None):
         "A class-based line/cell magic"
 
 
 class Awkward(object):
     def __getattr__(self, name):
         raise Exception(name)
 
 
 def check_calltip(obj, name, call, docstring):
     """Generic check pattern all calltip tests will use"""
     info = inspector.info(obj, name)
     call_line, ds = oinspect.call_tip(info)
     nt.assert_equal(call_line, call)
     nt.assert_equal(ds, docstring)
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 def test_calltip_class():
     check_calltip(Call, 'Call', 'Call(x, y=1)', Call.__init__.__doc__)
 
 
 def test_calltip_instance():
     c = Call(1)
     check_calltip(c, 'c', 'c(*a, **kw)', c.__call__.__doc__)
 
 
 def test_calltip_method():
     c = Call(1)
     check_calltip(c.method, 'c.method', 'c.method(x, z=2)', c.method.__doc__)
 
 
 def test_calltip_function():
     check_calltip(f, 'f', 'f(x, y=2, *a, **kw)', f.__doc__)
 
 
 def test_calltip_function2():
     check_calltip(g, 'g', 'g(y, z=3, *a, **kw)', '<no docstring>')
 
 
 def test_calltip_builtin():
     check_calltip(sum, 'sum', None, sum.__doc__)
 
 
 def test_calltip_line_magic():
     check_calltip(lmagic, 'lmagic', 'lmagic(line)', "A line magic")
 
         
 def test_calltip_cell_magic():
     check_calltip(cmagic, 'cmagic', 'cmagic(line, cell)', "A cell magic")
 
         
 def test_calltip_line_cell_magic():
     check_calltip(lcmagic, 'lcmagic', 'lcmagic(line, cell=None)', 
                   "A line/cell magic")
         
 
 def test_class_magics():
     cm = SimpleMagics(ip)
     ip.register_magics(cm)
     check_calltip(cm.Clmagic, 'Clmagic', 'Clmagic(cline)',
                   "A class-based line magic")
     check_calltip(cm.Ccmagic, 'Ccmagic', 'Ccmagic(cline, ccell)',
                   "A class-based cell magic")
     check_calltip(cm.Clcmagic, 'Clcmagic', 'Clcmagic(cline, ccell=None)',
                   "A class-based line/cell magic")
     
 
 def test_info():
     "Check that Inspector.info fills out various fields as expected."
     i = inspector.info(Call, oname='Call')
     nt.assert_equal(i['type_name'], 'type')
     expted_class = str(type(type))  # <class 'type'> (Python 3) or <type 'type'>
     nt.assert_equal(i['base_class'], expted_class)
     nt.assert_equal(i['string_form'], "<class 'IPython.core.tests.test_oinspect.Call'>")
     fname = __file__
     if fname.endswith(".pyc"):
         fname = fname[:-1]
     # case-insensitive comparison needed on some filesystems
     # e.g. Windows:
     nt.assert_equal(i['file'].lower(), compress_user(fname.lower()))
     nt.assert_equal(i['definition'], None)
     nt.assert_equal(i['docstring'], Call.__doc__)
     nt.assert_equal(i['source'], None)
     nt.assert_true(i['isclass'])
     nt.assert_equal(i['init_definition'], "Call(self, x, y=1)\n")
     nt.assert_equal(i['init_docstring'], Call.__init__.__doc__)
 
     i = inspector.info(Call, detail_level=1)
     nt.assert_not_equal(i['source'], None)
     nt.assert_equal(i['docstring'], None)
 
     c = Call(1)
     c.__doc__ = "Modified instance docstring"
     i = inspector.info(c)
     nt.assert_equal(i['type_name'], 'Call')
     nt.assert_equal(i['docstring'], "Modified instance docstring")
     nt.assert_equal(i['class_docstring'], Call.__doc__)
     nt.assert_equal(i['init_docstring'], Call.__init__.__doc__)
     nt.assert_equal(i['call_docstring'], Call.__call__.__doc__)
 
     # Test old-style classes, which for example may not have an __init__ method.
     if not py3compat.PY3:
         i = inspector.info(OldStyle)
         nt.assert_equal(i['type_name'], 'classobj')
 
         i = inspector.info(OldStyle())
         nt.assert_equal(i['type_name'], 'instance')
         nt.assert_equal(i['docstring'], OldStyle.__doc__)
 
 def test_info_awkward():
     # Just test that this doesn't throw an error.
     i = inspector.info(Awkward())
 
 def test_calldef_none():
     # We should ignore __call__ for all of these.
     for obj in [f, SimpleClass().method, any, str.upper]:
         print(obj)
         i = inspector.info(obj)
         nt.assert_is(i['call_def'], None)
 
 if py3compat.PY3:
     exec("def f_kwarg(pos, *, kwonly): pass")
 
 @skipif(not py3compat.PY3)
 def test_definition_kwonlyargs():
     i = inspector.info(f_kwarg, oname='f_kwarg')  # analysis:ignore
     nt.assert_equal(i['definition'], "f_kwarg(pos, *, kwonly)\n")
 
 def test_getdoc():
     class A(object):
         """standard docstring"""
         pass
     
     class B(object):
         """standard docstring"""
         def getdoc(self):
             return "custom docstring"
     
     class C(object):
         """standard docstring"""
         def getdoc(self):
             return None
     
     a = A()
     b = B()
     c = C()
     
     nt.assert_equal(oinspect.getdoc(a), "standard docstring")
     nt.assert_equal(oinspect.getdoc(b), "custom docstring")
     nt.assert_equal(oinspect.getdoc(c), "standard docstring")
 
 
 def test_empty_property_has_no_source():
     i = inspector.info(property(), detail_level=1)
     nt.assert_is(i['source'], None)
 
 
 def test_property_sources():
     import zlib
 
     class A(object):
         @property
         def foo(self):
             return 'bar'
 
         foo = foo.setter(lambda self, v: setattr(self, 'bar', v))
 
         id = property(id)
         compress = property(zlib.compress)
 
     i = inspector.info(A.foo, detail_level=1)
     nt.assert_in('def foo(self):', i['source'])
     nt.assert_in('lambda self, v:', i['source'])
 
     i = inspector.info(A.id, detail_level=1)
     nt.assert_in('fget = <function id>', i['source'])
 
     i = inspector.info(A.compress, detail_level=1)
     nt.assert_in('fget = <function zlib.compress>', i['source'])
 
 
 def test_property_docstring_is_in_info_for_detail_level_0():
     class A(object):
         @property
         def foobar():
             """This is `foobar` property."""
             pass
 
     ip.user_ns['a_obj'] = A()
     nt.assert_equals(
         'This is `foobar` property.',
         ip.object_inspect('a_obj.foobar', detail_level=0)['docstring'])
 
     ip.user_ns['a_cls'] = A
     nt.assert_equals(
         'This is `foobar` property.',
         ip.object_inspect('a_cls.foobar', detail_level=0)['docstring'])
 
 
 def test_pdef():
     # See gh-1914
     def foo(): pass
     inspector.pdef(foo, 'foo')
 
 def test_pinfo_nonascii():
     # See gh-1177
     from . import nonascii2
     ip.user_ns['nonascii2'] = nonascii2
     ip._inspect('pinfo', 'nonascii2', detail_level=1)
 
 def test_pinfo_magic():
     with AssertPrints('Docstring:'):
         ip._inspect('pinfo', 'lsmagic', detail_level=0)
 
     with AssertPrints('Source:'):
         ip._inspect('pinfo', 'lsmagic', detail_level=1)

IPython/parallel/client/asyncresult.py

@@ -1,703 +1,703 b''
 """AsyncResult objects for the client"""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import sys
 import time
 from datetime import datetime
 
 from zmq import MessageTracker
 
 from IPython.core.display import clear_output, display, display_pretty
-from IPython.external.decorator import decorator
+from decorator import decorator
 from IPython.parallel import error
 from IPython.utils.py3compat import string_types
 
 
 def _raw_text(s):
     display_pretty(s, raw=True)
 
 
 # global empty tracker that's always done:
 finished_tracker = MessageTracker()
 
 @decorator
 def check_ready(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.wait(0)
     if not self._ready:
         raise error.TimeoutError("result not ready")
     return f(self, *args, **kwargs)
 
 class AsyncResult(object):
     """Class for representing results of non-blocking calls.
 
     Provides the same interface as :py:class:`multiprocessing.pool.AsyncResult`.
     """
 
     msg_ids = None
     _targets = None
     _tracker = None
     _single_result = False
     owner = False,
 
     def __init__(self, client, msg_ids, fname='unknown', targets=None, tracker=None,
         owner=False,
     ):
         if isinstance(msg_ids, string_types):
             # always a list
             msg_ids = [msg_ids]
             self._single_result = True
         else:
             self._single_result = False
         if tracker is None:
             # default to always done
             tracker = finished_tracker
         self._client = client
         self.msg_ids = msg_ids
         self._fname=fname
         self._targets = targets
         self._tracker = tracker
         self.owner = owner
         
         self._ready = False
         self._outputs_ready = False
         self._success = None
         self._metadata = [self._client.metadata[id] for id in self.msg_ids]
 
     def __repr__(self):
         if self._ready:
             return "<%s: finished>"%(self.__class__.__name__)
         else:
             return "<%s: %s>"%(self.__class__.__name__,self._fname)
 
 
     def _reconstruct_result(self, res):
         """Reconstruct our result from actual result list (always a list)
 
         Override me in subclasses for turning a list of results
         into the expected form.
         """
         if self._single_result:
             return res[0]
         else:
             return res
 
     def get(self, timeout=-1):
         """Return the result when it arrives.
 
         If `timeout` is not ``None`` and the result does not arrive within
         `timeout` seconds then ``TimeoutError`` is raised. If the
         remote call raised an exception then that exception will be reraised
         by get() inside a `RemoteError`.
         """
         if not self.ready():
             self.wait(timeout)
 
         if self._ready:
             if self._success:
                 return self._result
             else:
                 raise self._exception
         else:
             raise error.TimeoutError("Result not ready.")
 
     def _check_ready(self):
         if not self.ready():
             raise error.TimeoutError("Result not ready.")
     
     def ready(self):
         """Return whether the call has completed."""
         if not self._ready:
             self.wait(0)
         elif not self._outputs_ready:
             self._wait_for_outputs(0)
         
         return self._ready
 
     def wait(self, timeout=-1):
         """Wait until the result is available or until `timeout` seconds pass.
 
         This method always returns None.
         """
         if self._ready:
             self._wait_for_outputs(timeout)
             return
         self._ready = self._client.wait(self.msg_ids, timeout)
         if self._ready:
             try:
                 results = list(map(self._client.results.get, self.msg_ids))
                 self._result = results
                 if self._single_result:
                     r = results[0]
                     if isinstance(r, Exception):
                         raise r
                 else:
                     results = error.collect_exceptions(results, self._fname)
                 self._result = self._reconstruct_result(results)
             except Exception as e:
                 self._exception = e
                 self._success = False
             else:
                 self._success = True
             finally:
                 if timeout is None or timeout < 0:
                     # cutoff infinite wait at 10s
                     timeout = 10
                 self._wait_for_outputs(timeout)
                 
                 if self.owner:
                     
                     self._metadata = [self._client.metadata.pop(mid) for mid in self.msg_ids]
                     [self._client.results.pop(mid) for mid in self.msg_ids]
                 
 
 
     def successful(self):
         """Return whether the call completed without raising an exception.
 
         Will raise ``AssertionError`` if the result is not ready.
         """
         assert self.ready()
         return self._success
 
     #----------------------------------------------------------------
     # Extra methods not in mp.pool.AsyncResult
     #----------------------------------------------------------------
 
     def get_dict(self, timeout=-1):
         """Get the results as a dict, keyed by engine_id.
 
         timeout behavior is described in `get()`.
         """
 
         results = self.get(timeout)
         if self._single_result:
             results = [results]
         engine_ids = [ md['engine_id'] for md in self._metadata ]
         
         
         rdict = {}
         for engine_id, result in zip(engine_ids, results):
             if engine_id in rdict:
                 raise ValueError("Cannot build dict, %i jobs ran on engine #%i" % (
                     engine_ids.count(engine_id), engine_id)
                 )
             else:
                 rdict[engine_id] = result
 
         return rdict
 
     @property
     def result(self):
         """result property wrapper for `get(timeout=-1)`."""
         return self.get()
 
     # abbreviated alias:
     r = result
 
     @property
     def metadata(self):
         """property for accessing execution metadata."""
         if self._single_result:
             return self._metadata[0]
         else:
             return self._metadata
 
     @property
     def result_dict(self):
         """result property as a dict."""
         return self.get_dict()
 
     def __dict__(self):
         return self.get_dict(0)
 
     def abort(self):
         """abort my tasks."""
         assert not self.ready(), "Can't abort, I am already done!"
         return self._client.abort(self.msg_ids, targets=self._targets, block=True)
 
     @property
     def sent(self):
         """check whether my messages have been sent."""
         return self._tracker.done
 
     def wait_for_send(self, timeout=-1):
         """wait for pyzmq send to complete.
 
         This is necessary when sending arrays that you intend to edit in-place.
         `timeout` is in seconds, and will raise TimeoutError if it is reached
         before the send completes.
         """
         return self._tracker.wait(timeout)
 
     #-------------------------------------
     # dict-access
     #-------------------------------------
 
     def __getitem__(self, key):
         """getitem returns result value(s) if keyed by int/slice, or metadata if key is str.
         """
         if isinstance(key, int):
             self._check_ready()
             return error.collect_exceptions([self._result[key]], self._fname)[0]
         elif isinstance(key, slice):
             self._check_ready()
             return error.collect_exceptions(self._result[key], self._fname)
         elif isinstance(key, string_types):
             # metadata proxy *does not* require that results are done
             self.wait(0)
             values = [ md[key] for md in self._metadata ]
             if self._single_result:
                 return values[0]
             else:
                 return values
         else:
             raise TypeError("Invalid key type %r, must be 'int','slice', or 'str'"%type(key))
 
     def __getattr__(self, key):
         """getattr maps to getitem for convenient attr access to metadata."""
         try:
             return self.__getitem__(key)
         except (error.TimeoutError, KeyError):
             raise AttributeError("%r object has no attribute %r"%(
                     self.__class__.__name__, key))
 
     # asynchronous iterator:
     def __iter__(self):
         if self._single_result:
             raise TypeError("AsyncResults with a single result are not iterable.")
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             # wait for each result individually
             for msg_id in self.msg_ids:
                 ar = AsyncResult(self._client, msg_id, self._fname)
                 yield ar.get()
         else:
             # already done
             for r in rlist:
                 yield r
     
     def __len__(self):
         return len(self.msg_ids)
     
     #-------------------------------------
     # Sugar methods and attributes
     #-------------------------------------
     
     def timedelta(self, start, end, start_key=min, end_key=max):
         """compute the difference between two sets of timestamps
         
         The default behavior is to use the earliest of the first
         and the latest of the second list, but this can be changed
         by passing a different
         
         Parameters
         ----------
         
         start : one or more datetime objects (e.g. ar.submitted)
         end : one or more datetime objects (e.g. ar.received)
         start_key : callable
             Function to call on `start` to extract the relevant
             entry [defalt: min]
         end_key : callable
             Function to call on `end` to extract the relevant
             entry [default: max]
         
         Returns
         -------
         
         dt : float
             The time elapsed (in seconds) between the two selected timestamps.
         """
         if not isinstance(start, datetime):
             # handle single_result AsyncResults, where ar.stamp is single object,
             # not a list
             start = start_key(start)
         if not isinstance(end, datetime):
             # handle single_result AsyncResults, where ar.stamp is single object,
             # not a list
             end = end_key(end)
         return (end - start).total_seconds()
         
     @property
     def progress(self):
         """the number of tasks which have been completed at this point.
         
         Fractional progress would be given by 1.0 * ar.progress / len(ar)
         """
         self.wait(0)
         return len(self) - len(set(self.msg_ids).intersection(self._client.outstanding))
     
     @property
     def elapsed(self):
         """elapsed time since initial submission"""
         if self.ready():
             return self.wall_time
         
         now = submitted = datetime.now()
         for msg_id in self.msg_ids:
             if msg_id in self._client.metadata:
                 stamp = self._client.metadata[msg_id]['submitted']
                 if stamp and stamp < submitted:
                     submitted = stamp
         return (now-submitted).total_seconds()
     
     @property
     @check_ready
     def serial_time(self):
         """serial computation time of a parallel calculation
         
         Computed as the sum of (completed-started) of each task
         """
         t = 0
         for md in self._metadata:
             t += (md['completed'] - md['started']).total_seconds()
         return t
     
     @property
     @check_ready
     def wall_time(self):
         """actual computation time of a parallel calculation
         
         Computed as the time between the latest `received` stamp
         and the earliest `submitted`.
         
         Only reliable if Client was spinning/waiting when the task finished, because
         the `received` timestamp is created when a result is pulled off of the zmq queue,
         which happens as a result of `client.spin()`.
         
         For similar comparison of other timestamp pairs, check out AsyncResult.timedelta.
         
         """
         return self.timedelta(self.submitted, self.received)
     
     def wait_interactive(self, interval=1., timeout=-1):
         """interactive wait, printing progress at regular intervals"""
         if timeout is None:
             timeout = -1
         N = len(self)
         tic = time.time()
         while not self.ready() and (timeout < 0 or time.time() - tic <= timeout):
             self.wait(interval)
             clear_output(wait=True)
             print("%4i/%i tasks finished after %4i s" % (self.progress, N, self.elapsed), end="")
             sys.stdout.flush()
         print()
         print("done")
     
     def _republish_displaypub(self, content, eid):
         """republish individual displaypub content dicts"""
         try:
             ip = get_ipython()
         except NameError:
             # displaypub is meaningless outside IPython
             return
         md = content['metadata'] or {}
         md['engine'] = eid
         ip.display_pub.publish(data=content['data'], metadata=md)
     
     def _display_stream(self, text, prefix='', file=None):
         if not text:
             # nothing to display
             return
         if file is None:
             file = sys.stdout
         end = '' if text.endswith('\n') else '\n'
         
         multiline = text.count('\n') > int(text.endswith('\n'))
         if prefix and multiline and not text.startswith('\n'):
             prefix = prefix + '\n'
         print("%s%s" % (prefix, text), file=file, end=end)
         
     
     def _display_single_result(self):
         self._display_stream(self.stdout)
         self._display_stream(self.stderr, file=sys.stderr)
         
         try:
             get_ipython()
         except NameError:
             # displaypub is meaningless outside IPython
             return
         
         for output in self.outputs:
             self._republish_displaypub(output, self.engine_id)
         
         if self.execute_result is not None:
             display(self.get())
     
     def _wait_for_outputs(self, timeout=-1):
         """wait for the 'status=idle' message that indicates we have all outputs
         """
         if self._outputs_ready or not self._success:
             # don't wait on errors
             return
         
         # cast None to -1 for infinite timeout
         if timeout is None:
             timeout = -1
         
         tic = time.time()
         while True:
             self._client._flush_iopub(self._client._iopub_socket)
             self._outputs_ready = all(md['outputs_ready']
                                       for md in self._metadata)
             if self._outputs_ready or \
                (timeout >= 0 and time.time() > tic + timeout):
                 break
             time.sleep(0.01)
     
     @check_ready
     def display_outputs(self, groupby="type"):
         """republish the outputs of the computation
         
         Parameters
         ----------
         
         groupby : str [default: type]
             if 'type':
                 Group outputs by type (show all stdout, then all stderr, etc.):
                 
                 [stdout:1] foo
                 [stdout:2] foo
                 [stderr:1] bar
                 [stderr:2] bar
             if 'engine':
                 Display outputs for each engine before moving on to the next:
                 
                 [stdout:1] foo
                 [stderr:1] bar
                 [stdout:2] foo
                 [stderr:2] bar
                 
             if 'order':
                 Like 'type', but further collate individual displaypub
                 outputs.  This is meant for cases of each command producing
                 several plots, and you would like to see all of the first
                 plots together, then all of the second plots, and so on.
         """
         if self._single_result:
             self._display_single_result()
             return
         
         stdouts = self.stdout
         stderrs = self.stderr
         execute_results  = self.execute_result
         output_lists = self.outputs
         results = self.get()
         
         targets = self.engine_id
         
         if groupby == "engine":
             for eid,stdout,stderr,outputs,r,execute_result in zip(
                     targets, stdouts, stderrs, output_lists, results, execute_results
                 ):
                 self._display_stream(stdout, '[stdout:%i] ' % eid)
                 self._display_stream(stderr, '[stderr:%i] ' % eid, file=sys.stderr)
                 
                 try:
                     get_ipython()
                 except NameError:
                     # displaypub is meaningless outside IPython
                     return 
                 
                 if outputs or execute_result is not None:
                     _raw_text('[output:%i]' % eid)
                 
                 for output in outputs:
                     self._republish_displaypub(output, eid)
                 
                 if execute_result is not None:
                     display(r)
         
         elif groupby in ('type', 'order'):
             # republish stdout:
             for eid,stdout in zip(targets, stdouts):
                 self._display_stream(stdout, '[stdout:%i] ' % eid)
         
             # republish stderr:
             for eid,stderr in zip(targets, stderrs):
                 self._display_stream(stderr, '[stderr:%i] ' % eid, file=sys.stderr)
         
             try:
                 get_ipython()
             except NameError:
                 # displaypub is meaningless outside IPython
                 return
             
             if groupby == 'order':
                 output_dict = dict((eid, outputs) for eid,outputs in zip(targets, output_lists))
                 N = max(len(outputs) for outputs in output_lists)
                 for i in range(N):
                     for eid in targets:
                         outputs = output_dict[eid]
                         if len(outputs) >= N:
                             _raw_text('[output:%i]' % eid)
                             self._republish_displaypub(outputs[i], eid)
             else:
                 # republish displaypub output
                 for eid,outputs in zip(targets, output_lists):
                     if outputs:
                         _raw_text('[output:%i]' % eid)
                     for output in outputs:
                         self._republish_displaypub(output, eid)
         
             # finally, add execute_result:
             for eid,r,execute_result in zip(targets, results, execute_results):
                 if execute_result is not None:
                     display(r)
         
         else:
             raise ValueError("groupby must be one of 'type', 'engine', 'collate', not %r" % groupby)
         
         
 
 
 class AsyncMapResult(AsyncResult):
     """Class for representing results of non-blocking gathers.
 
     This will properly reconstruct the gather.
     
     This class is iterable at any time, and will wait on results as they come.
     
     If ordered=False, then the first results to arrive will come first, otherwise
     results will be yielded in the order they were submitted.
     
     """
 
     def __init__(self, client, msg_ids, mapObject, fname='', ordered=True):
         AsyncResult.__init__(self, client, msg_ids, fname=fname)
         self._mapObject = mapObject
         self._single_result = False
         self.ordered = ordered
 
     def _reconstruct_result(self, res):
         """Perform the gather on the actual results."""
         return self._mapObject.joinPartitions(res)
 
     # asynchronous iterator:
     def __iter__(self):
         it = self._ordered_iter if self.ordered else self._unordered_iter
         for r in it():
             yield r
 
     # asynchronous ordered iterator:
     def _ordered_iter(self):
         """iterator for results *as they arrive*, preserving submission order."""
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             # wait for each result individually
             for msg_id in self.msg_ids:
                 ar = AsyncResult(self._client, msg_id, self._fname)
                 rlist = ar.get()
                 try:
                     for r in rlist:
                         yield r
                 except TypeError:
                     # flattened, not a list
                     # this could get broken by flattened data that returns iterables
                     # but most calls to map do not expose the `flatten` argument
                     yield rlist
         else:
             # already done
             for r in rlist:
                 yield r
 
     # asynchronous unordered iterator:
     def _unordered_iter(self):
         """iterator for results *as they arrive*, on FCFS basis, ignoring submission order."""
         try:
             rlist = self.get(0)
         except error.TimeoutError:
             pending = set(self.msg_ids)
             while pending:
                 try:
                     self._client.wait(pending, 1e-3)
                 except error.TimeoutError:
                     # ignore timeout error, because that only means
                     # *some* jobs are outstanding
                     pass
                 # update ready set with those no longer outstanding:
                 ready = pending.difference(self._client.outstanding)
                 # update pending to exclude those that are finished
                 pending = pending.difference(ready)
                 while ready:
                     msg_id = ready.pop()
                     ar = AsyncResult(self._client, msg_id, self._fname)
                     rlist = ar.get()
                     try:
                         for r in rlist:
                             yield r
                     except TypeError:
                         # flattened, not a list
                         # this could get broken by flattened data that returns iterables
                         # but most calls to map do not expose the `flatten` argument
                         yield rlist
         else:
             # already done
             for r in rlist:
                 yield r
 
 
 class AsyncHubResult(AsyncResult):
     """Class to wrap pending results that must be requested from the Hub.
 
     Note that waiting/polling on these objects requires polling the Hubover the network,
     so use `AsyncHubResult.wait()` sparingly.
     """
 
     def _wait_for_outputs(self, timeout=-1):
         """no-op, because HubResults are never incomplete"""
         self._outputs_ready = True
     
     def wait(self, timeout=-1):
         """wait for result to complete."""
         start = time.time()
         if self._ready:
             return
         local_ids = [m for m in self.msg_ids if m in self._client.outstanding]
         local_ready = self._client.wait(local_ids, timeout)
         if local_ready:
             remote_ids = [m for m in self.msg_ids if m not in self._client.results]
             if not remote_ids:
                 self._ready = True
             else:
                 rdict = self._client.result_status(remote_ids, status_only=False)
                 pending = rdict['pending']
                 while pending and (timeout < 0 or time.time() < start+timeout):
                     rdict = self._client.result_status(remote_ids, status_only=False)
                     pending = rdict['pending']
                     if pending:
                         time.sleep(0.1)
                 if not pending:
                     self._ready = True
         if self._ready:
             try:
                 results = list(map(self._client.results.get, self.msg_ids))
                 self._result = results
                 if self._single_result:
                     r = results[0]
                     if isinstance(r, Exception):
                         raise r
                 else:
                     results = error.collect_exceptions(results, self._fname)
                 self._result = self._reconstruct_result(results)
             except Exception as e:
                 self._exception = e
                 self._success = False
             else:
                 self._success = True
             finally:
                 self._metadata = [self._client.metadata[mid] for mid in self.msg_ids]
                 if self.owner:
                     [self._client.metadata.pop(mid) for mid in self.msg_ids]
                     [self._client.results.pop(mid) for mid in self.msg_ids]
             
 
 __all__ = ['AsyncResult', 'AsyncMapResult', 'AsyncHubResult']

IPython/parallel/client/client.py

@@ -1,1893 +1,1893 b''
 """A semi-synchronous Client for IPython parallel"""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import os
 import json
 import sys
 from threading import Thread, Event
 import time
 import warnings
 from datetime import datetime
 from getpass import getpass
 from pprint import pprint
 
 pjoin = os.path.join
 
 import zmq
 
 from IPython.config.configurable import MultipleInstanceError
 from IPython.core.application import BaseIPythonApplication
 from IPython.core.profiledir import ProfileDir, ProfileDirError
 
 from IPython.utils.capture import RichOutput
 from IPython.utils.coloransi import TermColors
 from IPython.utils.jsonutil import rekey, extract_dates, parse_date
 from IPython.utils.localinterfaces import localhost, is_local_ip
 from IPython.utils.path import get_ipython_dir, compress_user
 from IPython.utils.py3compat import cast_bytes, string_types, xrange, iteritems
 from IPython.utils.traitlets import (HasTraits, Integer, Instance, Unicode,
                                     Dict, List, Bool, Set, Any)
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 from IPython.parallel import Reference
 from IPython.parallel import error
 from IPython.parallel import util
 
 from IPython.kernel.zmq.session import Session, Message
 from IPython.kernel.zmq import serialize
 
 from .asyncresult import AsyncResult, AsyncHubResult
 from .view import DirectView, LoadBalancedView
 
 #--------------------------------------------------------------------------
 # Decorators for Client methods
 #--------------------------------------------------------------------------
 
 
 @decorator
 def spin_first(f, self, *args, **kwargs):
     """Call spin() to sync state prior to calling the method."""
     self.spin()
     return f(self, *args, **kwargs)
 
 
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 _no_connection_file_msg = """
 Failed to connect because no Controller could be found.
 Please double-check your profile and ensure that a cluster is running.
 """
 
 class ExecuteReply(RichOutput):
     """wrapper for finished Execute results"""
     def __init__(self, msg_id, content, metadata):
         self.msg_id = msg_id
         self._content = content
         self.execution_count = content['execution_count']
         self.metadata = metadata
     
     # RichOutput overrides
     
     @property
     def source(self):
         execute_result = self.metadata['execute_result']
         if execute_result:
             return execute_result.get('source', '')
     
     @property
     def data(self):
         execute_result = self.metadata['execute_result']
         if execute_result:
             return execute_result.get('data', {})
     
     @property
     def _metadata(self):
         execute_result = self.metadata['execute_result']
         if execute_result:
             return execute_result.get('metadata', {})
     
     def display(self):
         from IPython.display import publish_display_data
         publish_display_data(self.data, self.metadata)
     
     def _repr_mime_(self, mime):
         if mime not in self.data:
             return
         data = self.data[mime]
         if mime in self._metadata:
             return data, self._metadata[mime]
         else:
             return data
     
     def __getitem__(self, key):
         return self.metadata[key]
     
     def __getattr__(self, key):
         if key not in self.metadata:
             raise AttributeError(key)
         return self.metadata[key]
     
     def __repr__(self):
         execute_result = self.metadata['execute_result'] or {'data':{}}
         text_out = execute_result['data'].get('text/plain', '')
         if len(text_out) > 32:
             text_out = text_out[:29] + '...'
         
         return "<ExecuteReply[%i]: %s>" % (self.execution_count, text_out)
     
     def _repr_pretty_(self, p, cycle):
         execute_result = self.metadata['execute_result'] or {'data':{}}
         text_out = execute_result['data'].get('text/plain', '')
         
         if not text_out:
             return
         
         try:
             ip = get_ipython()
         except NameError:
             colors = "NoColor"
         else:
             colors = ip.colors
         
         if colors == "NoColor":
             out = normal = ""
         else:
             out = TermColors.Red
             normal = TermColors.Normal
         
         if '\n' in text_out and not text_out.startswith('\n'):
             # add newline for multiline reprs
             text_out = '\n' + text_out
         
         p.text(
             out + u'Out[%i:%i]: ' % (
                 self.metadata['engine_id'], self.execution_count
             ) + normal + text_out
         )
 
 
 class Metadata(dict):
     """Subclass of dict for initializing metadata values.
 
     Attribute access works on keys.
 
     These objects have a strict set of keys - errors will raise if you try
     to add new keys.
     """
     def __init__(self, *args, **kwargs):
         dict.__init__(self)
         md = {'msg_id' : None,
               'submitted' : None,
               'started' : None,
               'completed' : None,
               'received' : None,
               'engine_uuid' : None,
               'engine_id' : None,
               'follow' : None,
               'after' : None,
               'status' : None,
 
               'execute_input' : None,
               'execute_result' : None,
               'error' : None,
               'stdout' : '',
               'stderr' : '',
               'outputs' : [],
               'data': {},
               'outputs_ready' : False,
             }
         self.update(md)
         self.update(dict(*args, **kwargs))
 
     def __getattr__(self, key):
         """getattr aliased to getitem"""
         if key in self:
             return self[key]
         else:
             raise AttributeError(key)
 
     def __setattr__(self, key, value):
         """setattr aliased to setitem, with strict"""
         if key in self:
             self[key] = value
         else:
             raise AttributeError(key)
 
     def __setitem__(self, key, value):
         """strict static key enforcement"""
         if key in self:
             dict.__setitem__(self, key, value)
         else:
             raise KeyError(key)
 
 
 class Client(HasTraits):
     """A semi-synchronous client to the IPython ZMQ cluster
 
     Parameters
     ----------
 
     url_file : str/unicode; path to ipcontroller-client.json
         This JSON file should contain all the information needed to connect to a cluster,
         and is likely the only argument needed.
         Connection information for the Hub's registration.  If a json connector
         file is given, then likely no further configuration is necessary.
         [Default: use profile]
     profile : bytes
         The name of the Cluster profile to be used to find connector information.
         If run from an IPython application, the default profile will be the same
         as the running application, otherwise it will be 'default'.
     cluster_id : str
         String id to added to runtime files, to prevent name collisions when using
         multiple clusters with a single profile simultaneously.
         When set, will look for files named like: 'ipcontroller-<cluster_id>-client.json'
         Since this is text inserted into filenames, typical recommendations apply:
         Simple character strings are ideal, and spaces are not recommended (but
         should generally work)
     context : zmq.Context
         Pass an existing zmq.Context instance, otherwise the client will create its own.
     debug : bool
         flag for lots of message printing for debug purposes
     timeout : int/float
         time (in seconds) to wait for connection replies from the Hub
         [Default: 10]
 
     #-------------- session related args ----------------
 
     config : Config object
         If specified, this will be relayed to the Session for configuration
     username : str
         set username for the session object
 
     #-------------- ssh related args ----------------
     # These are args for configuring the ssh tunnel to be used
     # credentials are used to forward connections over ssh to the Controller
     # Note that the ip given in `addr` needs to be relative to sshserver
     # The most basic case is to leave addr as pointing to localhost (127.0.0.1),
     # and set sshserver as the same machine the Controller is on. However,
     # the only requirement is that sshserver is able to see the Controller
     # (i.e. is within the same trusted network).
 
     sshserver : str
         A string of the form passed to ssh, i.e. 'server.tld' or '[e-mail]:port'
         If keyfile or password is specified, and this is not, it will default to
         the ip given in addr.
     sshkey : str; path to ssh private key file
         This specifies a key to be used in ssh login, default None.
         Regular default ssh keys will be used without specifying this argument.
     password : str
         Your ssh password to sshserver. Note that if this is left None,
         you will be prompted for it if passwordless key based login is unavailable.
     paramiko : bool
         flag for whether to use paramiko instead of shell ssh for tunneling.
         [default: True on win32, False else]
 
 
     Attributes
     ----------
 
     ids : list of int engine IDs
         requesting the ids attribute always synchronizes
         the registration state. To request ids without synchronization,
         use semi-private _ids attributes.
 
     history : list of msg_ids
         a list of msg_ids, keeping track of all the execution
         messages you have submitted in order.
 
     outstanding : set of msg_ids
         a set of msg_ids that have been submitted, but whose
         results have not yet been received.
 
     results : dict
         a dict of all our results, keyed by msg_id
 
     block : bool
         determines default behavior when block not specified
         in execution methods
 
     Methods
     -------
 
     spin
         flushes incoming results and registration state changes
         control methods spin, and requesting `ids` also ensures up to date
 
     wait
         wait on one or more msg_ids
 
     execution methods
         apply
         legacy: execute, run
 
     data movement
         push, pull, scatter, gather
 
     query methods
         queue_status, get_result, purge, result_status
 
     control methods
         abort, shutdown
 
     """
 
 
     block = Bool(False)
     outstanding = Set()
     results = Instance('collections.defaultdict', (dict,))
     metadata = Instance('collections.defaultdict', (Metadata,))
     history = List()
     debug = Bool(False)
     _spin_thread = Any()
     _stop_spinning = Any()
 
     profile=Unicode()
     def _profile_default(self):
         if BaseIPythonApplication.initialized():
             # an IPython app *might* be running, try to get its profile
             try:
                 return BaseIPythonApplication.instance().profile
             except (AttributeError, MultipleInstanceError):
                 # could be a *different* subclass of config.Application,
                 # which would raise one of these two errors.
                 return u'default'
         else:
             return u'default'
 
 
     _outstanding_dict = Instance('collections.defaultdict', (set,))
     _ids = List()
     _connected=Bool(False)
     _ssh=Bool(False)
     _context = Instance('zmq.Context')
     _config = Dict()
     _engines=Instance(util.ReverseDict, (), {})
     # _hub_socket=Instance('zmq.Socket')
     _query_socket=Instance('zmq.Socket')
     _control_socket=Instance('zmq.Socket')
     _iopub_socket=Instance('zmq.Socket')
     _notification_socket=Instance('zmq.Socket')
     _mux_socket=Instance('zmq.Socket')
     _task_socket=Instance('zmq.Socket')
     _task_scheme=Unicode()
     _closed = False
     _ignored_control_replies=Integer(0)
     _ignored_hub_replies=Integer(0)
 
     def __new__(self, *args, **kw):
         # don't raise on positional args
         return HasTraits.__new__(self, **kw)
 
     def __init__(self, url_file=None, profile=None, profile_dir=None, ipython_dir=None,
             context=None, debug=False,
             sshserver=None, sshkey=None, password=None, paramiko=None,
             timeout=10, cluster_id=None, **extra_args
             ):
         if profile:
             super(Client, self).__init__(debug=debug, profile=profile)
         else:
             super(Client, self).__init__(debug=debug)
         if context is None:
             context = zmq.Context.instance()
         self._context = context
         self._stop_spinning = Event()
         
         if 'url_or_file' in extra_args:
             url_file = extra_args['url_or_file']
             warnings.warn("url_or_file arg no longer supported, use url_file", DeprecationWarning)
         
         if url_file and util.is_url(url_file):
             raise ValueError("single urls cannot be specified, url-files must be used.")
 
         self._setup_profile_dir(self.profile, profile_dir, ipython_dir)
         
         no_file_msg = '\n'.join([
         "You have attempted to connect to an IPython Cluster but no Controller could be found.",
         "Please double-check your configuration and ensure that a cluster is running.",
         ])
         
         if self._cd is not None:
             if url_file is None:
                 if not cluster_id:
                     client_json = 'ipcontroller-client.json'
                 else:
                     client_json = 'ipcontroller-%s-client.json' % cluster_id
                 url_file = pjoin(self._cd.security_dir, client_json)
                 if not os.path.exists(url_file):
                     msg = '\n'.join([
                         "Connection file %r not found." % compress_user(url_file),
                         no_file_msg,
                     ])
                     raise IOError(msg)
         if url_file is None:
             raise IOError(no_file_msg)
         
         if not os.path.exists(url_file):
             # Connection file explicitly specified, but not found
             raise IOError("Connection file %r not found. Is a controller running?" % \
                 compress_user(url_file)
             )
         
         with open(url_file) as f:
             cfg = json.load(f)
         
         self._task_scheme = cfg['task_scheme']
 
         # sync defaults from args, json:
         if sshserver:
             cfg['ssh'] = sshserver
 
         location = cfg.setdefault('location', None)
         
         proto,addr = cfg['interface'].split('://')
         addr = util.disambiguate_ip_address(addr, location)
         cfg['interface'] = "%s://%s" % (proto, addr)
         
         # turn interface,port into full urls:
         for key in ('control', 'task', 'mux', 'iopub', 'notification', 'registration'):
             cfg[key] = cfg['interface'] + ':%i' % cfg[key]
         
         url = cfg['registration']
         
         if location is not None and addr == localhost():
             # location specified, and connection is expected to be local
             if not is_local_ip(location) and not sshserver:
                 # load ssh from JSON *only* if the controller is not on
                 # this machine
                 sshserver=cfg['ssh']
             if not is_local_ip(location) and not sshserver:
                 # warn if no ssh specified, but SSH is probably needed
                 # This is only a warning, because the most likely cause
                 # is a local Controller on a laptop whose IP is dynamic
                 warnings.warn("""
             Controller appears to be listening on localhost, but not on this machine.
             If this is true, you should specify Client(...,sshserver='you@%s')
             or instruct your controller to listen on an external IP."""%location,
                 RuntimeWarning)
         elif not sshserver:
             # otherwise sync with cfg
             sshserver = cfg['ssh']
 
         self._config = cfg
 
         self._ssh = bool(sshserver or sshkey or password)
         if self._ssh and sshserver is None:
             # default to ssh via localhost
             sshserver = addr
         if self._ssh and password is None:
             from zmq.ssh import tunnel
             if tunnel.try_passwordless_ssh(sshserver, sshkey, paramiko):
                 password=False
             else:
                 password = getpass("SSH Password for %s: "%sshserver)
         ssh_kwargs = dict(keyfile=sshkey, password=password, paramiko=paramiko)
 
         # configure and construct the session
         try:
             extra_args['packer'] = cfg['pack']
             extra_args['unpacker'] = cfg['unpack']
             extra_args['key'] = cast_bytes(cfg['key'])
             extra_args['signature_scheme'] = cfg['signature_scheme']
         except KeyError as exc:
             msg = '\n'.join([
                 "Connection file is invalid (missing '{}'), possibly from an old version of IPython.",
                 "If you are reusing connection files, remove them and start ipcontroller again."
             ])
             raise ValueError(msg.format(exc.message))
         
         self.session = Session(**extra_args)
 
         self._query_socket = self._context.socket(zmq.DEALER)
 
         if self._ssh:
             from zmq.ssh import tunnel
             tunnel.tunnel_connection(self._query_socket, cfg['registration'], sshserver, **ssh_kwargs)
         else:
             self._query_socket.connect(cfg['registration'])
 
         self.session.debug = self.debug
 
         self._notification_handlers = {'registration_notification' : self._register_engine,
                                     'unregistration_notification' : self._unregister_engine,
                                     'shutdown_notification' : lambda msg: self.close(),
                                     }
         self._queue_handlers = {'execute_reply' : self._handle_execute_reply,
                                 'apply_reply' : self._handle_apply_reply}
         
         try:
             self._connect(sshserver, ssh_kwargs, timeout)
         except:
             self.close(linger=0)
             raise
         
         # last step: setup magics, if we are in IPython:
         
         try:
             ip = get_ipython()
         except NameError:
             return
         else:
             if 'px' not in ip.magics_manager.magics:
                 # in IPython but we are the first Client.
                 # activate a default view for parallel magics.
                 self.activate()
 
     def __del__(self):
         """cleanup sockets, but _not_ context."""
         self.close()
 
     def _setup_profile_dir(self, profile, profile_dir, ipython_dir):
         if ipython_dir is None:
             ipython_dir = get_ipython_dir()
         if profile_dir is not None:
             try:
                 self._cd = ProfileDir.find_profile_dir(profile_dir)
                 return
             except ProfileDirError:
                 pass
         elif profile is not None:
             try:
                 self._cd = ProfileDir.find_profile_dir_by_name(
                     ipython_dir, profile)
                 return
             except ProfileDirError:
                 pass
         self._cd = None
 
     def _update_engines(self, engines):
         """Update our engines dict and _ids from a dict of the form: {id:uuid}."""
         for k,v in iteritems(engines):
             eid = int(k)
             if eid not in self._engines:
                 self._ids.append(eid)
             self._engines[eid] = v
         self._ids = sorted(self._ids)
         if sorted(self._engines.keys()) != list(range(len(self._engines))) and \
                         self._task_scheme == 'pure' and self._task_socket:
             self._stop_scheduling_tasks()
 
     def _stop_scheduling_tasks(self):
         """Stop scheduling tasks because an engine has been unregistered
         from a pure ZMQ scheduler.
         """
         self._task_socket.close()
         self._task_socket = None
         msg = "An engine has been unregistered, and we are using pure " +\
               "ZMQ task scheduling.  Task farming will be disabled."
         if self.outstanding:
             msg += " If you were running tasks when this happened, " +\
                    "some `outstanding` msg_ids may never resolve."
         warnings.warn(msg, RuntimeWarning)
 
     def _build_targets(self, targets):
         """Turn valid target IDs or 'all' into two lists:
         (int_ids, uuids).
         """
         if not self._ids:
             # flush notification socket if no engines yet, just in case
             if not self.ids:
                 raise error.NoEnginesRegistered("Can't build targets without any engines")
 
         if targets is None:
             targets = self._ids
         elif isinstance(targets, string_types):
             if targets.lower() == 'all':
                 targets = self._ids
             else:
                 raise TypeError("%r not valid str target, must be 'all'"%(targets))
         elif isinstance(targets, int):
             if targets < 0:
                 targets = self.ids[targets]
             if targets not in self._ids:
                 raise IndexError("No such engine: %i"%targets)
             targets = [targets]
 
         if isinstance(targets, slice):
             indices = list(range(len(self._ids))[targets])
             ids = self.ids
             targets = [ ids[i] for i in indices ]
 
         if not isinstance(targets, (tuple, list, xrange)):
             raise TypeError("targets by int/slice/collection of ints only, not %s"%(type(targets)))
 
         return [cast_bytes(self._engines[t]) for t in targets], list(targets)
 
     def _connect(self, sshserver, ssh_kwargs, timeout):
         """setup all our socket connections to the cluster. This is called from
         __init__."""
 
         # Maybe allow reconnecting?
         if self._connected:
             return
         self._connected=True
 
         def connect_socket(s, url):
             if self._ssh:
                 from zmq.ssh import tunnel
                 return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs)
             else:
                 return s.connect(url)
 
         self.session.send(self._query_socket, 'connection_request')
         # use Poller because zmq.select has wrong units in pyzmq 2.1.7
         poller = zmq.Poller()
         poller.register(self._query_socket, zmq.POLLIN)
         # poll expects milliseconds, timeout is seconds
         evts = poller.poll(timeout*1000)
         if not evts:
             raise error.TimeoutError("Hub connection request timed out")
         idents,msg = self.session.recv(self._query_socket,mode=0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         # self._config['registration'] = dict(content)
         cfg = self._config
         if content['status'] == 'ok':
             self._mux_socket = self._context.socket(zmq.DEALER)
             connect_socket(self._mux_socket, cfg['mux'])
 
             self._task_socket = self._context.socket(zmq.DEALER)
             connect_socket(self._task_socket, cfg['task'])
 
             self._notification_socket = self._context.socket(zmq.SUB)
             self._notification_socket.setsockopt(zmq.SUBSCRIBE, b'')
             connect_socket(self._notification_socket, cfg['notification'])
 
             self._control_socket = self._context.socket(zmq.DEALER)
             connect_socket(self._control_socket, cfg['control'])
 
             self._iopub_socket = self._context.socket(zmq.SUB)
             self._iopub_socket.setsockopt(zmq.SUBSCRIBE, b'')
             connect_socket(self._iopub_socket, cfg['iopub'])
 
             self._update_engines(dict(content['engines']))
         else:
             self._connected = False
             raise Exception("Failed to connect!")
 
     #--------------------------------------------------------------------------
     # handlers and callbacks for incoming messages
     #--------------------------------------------------------------------------
 
     def _unwrap_exception(self, content):
         """unwrap exception, and remap engine_id to int."""
         e = error.unwrap_exception(content)
         # print e.traceback
         if e.engine_info:
             e_uuid = e.engine_info['engine_uuid']
             eid = self._engines[e_uuid]
             e.engine_info['engine_id'] = eid
         return e
 
     def _extract_metadata(self, msg):
         header = msg['header']
         parent = msg['parent_header']
         msg_meta = msg['metadata']
         content = msg['content']
         md = {'msg_id' : parent['msg_id'],
               'received' : datetime.now(),
               'engine_uuid' : msg_meta.get('engine', None),
               'follow' : msg_meta.get('follow', []),
               'after' : msg_meta.get('after', []),
               'status' : content['status'],
             }
 
         if md['engine_uuid'] is not None:
             md['engine_id'] = self._engines.get(md['engine_uuid'], None)
 
         if 'date' in parent:
             md['submitted'] = parent['date']
         if 'started' in msg_meta:
             md['started'] = parse_date(msg_meta['started'])
         if 'date' in header:
             md['completed'] = header['date']
         return md
 
     def _register_engine(self, msg):
         """Register a new engine, and update our connection info."""
         content = msg['content']
         eid = content['id']
         d = {eid : content['uuid']}
         self._update_engines(d)
 
     def _unregister_engine(self, msg):
         """Unregister an engine that has died."""
         content = msg['content']
         eid = int(content['id'])
         if eid in self._ids:
             self._ids.remove(eid)
             uuid = self._engines.pop(eid)
 
             self._handle_stranded_msgs(eid, uuid)
 
         if self._task_socket and self._task_scheme == 'pure':
             self._stop_scheduling_tasks()
 
     def _handle_stranded_msgs(self, eid, uuid):
         """Handle messages known to be on an engine when the engine unregisters.
 
         It is possible that this will fire prematurely - that is, an engine will
         go down after completing a result, and the client will be notified
         of the unregistration and later receive the successful result.
         """
 
         outstanding = self._outstanding_dict[uuid]
 
         for msg_id in list(outstanding):
             if msg_id in self.results:
                 # we already
                 continue
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id))
             except:
                 content = error.wrap_exception()
             # build a fake message:
             msg = self.session.msg('apply_reply', content=content)
             msg['parent_header']['msg_id'] = msg_id
             msg['metadata']['engine'] = uuid
             self._handle_apply_reply(msg)
 
     def _handle_execute_reply(self, msg):
         """Save the reply to an execute_request into our results.
 
         execute messages are never actually used. apply is used instead.
         """
 
         parent = msg['parent_header']
         msg_id = parent['msg_id']
         if msg_id not in self.outstanding:
             if msg_id in self.history:
                 print("got stale result: %s"%msg_id)
             else:
                 print("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
 
         content = msg['content']
         header = msg['header']
 
         # construct metadata:
         md = self.metadata[msg_id]
         md.update(self._extract_metadata(msg))
         # is this redundant?
         self.metadata[msg_id] = md
         
         e_outstanding = self._outstanding_dict[md['engine_uuid']]
         if msg_id in e_outstanding:
             e_outstanding.remove(msg_id)
 
         # construct result:
         if content['status'] == 'ok':
             self.results[msg_id] = ExecuteReply(msg_id, content, md)
         elif content['status'] == 'aborted':
             self.results[msg_id] = error.TaskAborted(msg_id)
         elif content['status'] == 'resubmitted':
             # TODO: handle resubmission
             pass
         else:
             self.results[msg_id] = self._unwrap_exception(content)
 
     def _handle_apply_reply(self, msg):
         """Save the reply to an apply_request into our results."""
         parent = msg['parent_header']
         msg_id = parent['msg_id']
         if msg_id not in self.outstanding:
             if msg_id in self.history:
                 print("got stale result: %s"%msg_id)
                 print(self.results[msg_id])
                 print(msg)
             else:
                 print("got unknown result: %s"%msg_id)
         else:
             self.outstanding.remove(msg_id)
         content = msg['content']
         header = msg['header']
 
         # construct metadata:
         md = self.metadata[msg_id]
         md.update(self._extract_metadata(msg))
         # is this redundant?
         self.metadata[msg_id] = md
 
         e_outstanding = self._outstanding_dict[md['engine_uuid']]
         if msg_id in e_outstanding:
             e_outstanding.remove(msg_id)
 
         # construct result:
         if content['status'] == 'ok':
             self.results[msg_id] = serialize.deserialize_object(msg['buffers'])[0]
         elif content['status'] == 'aborted':
             self.results[msg_id] = error.TaskAborted(msg_id)
         elif content['status'] == 'resubmitted':
             # TODO: handle resubmission
             pass
         else:
             self.results[msg_id] = self._unwrap_exception(content)
 
     def _flush_notifications(self):
         """Flush notifications of engine registrations waiting
         in ZMQ queue."""
         idents,msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg_type = msg['header']['msg_type']
             handler = self._notification_handlers.get(msg_type, None)
             if handler is None:
                 raise Exception("Unhandled message type: %s" % msg_type)
             else:
                 handler(msg)
             idents,msg = self.session.recv(self._notification_socket, mode=zmq.NOBLOCK)
 
     def _flush_results(self, sock):
         """Flush task or queue results waiting in ZMQ queue."""
         idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             msg_type = msg['header']['msg_type']
             handler = self._queue_handlers.get(msg_type, None)
             if handler is None:
                 raise Exception("Unhandled message type: %s" % msg_type)
             else:
                 handler(msg)
             idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
 
     def _flush_control(self, sock):
         """Flush replies from the control channel waiting
         in the ZMQ queue.
 
         Currently: ignore them."""
         if self._ignored_control_replies <= 0:
             return
         idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             self._ignored_control_replies -= 1
             if self.debug:
                 pprint(msg)
             idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
 
     def _flush_ignored_control(self):
         """flush ignored control replies"""
         while self._ignored_control_replies > 0:
             self.session.recv(self._control_socket)
             self._ignored_control_replies -= 1
 
     def _flush_ignored_hub_replies(self):
         ident,msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)
         while msg is not None:
             ident,msg = self.session.recv(self._query_socket, mode=zmq.NOBLOCK)
 
     def _flush_iopub(self, sock):
         """Flush replies from the iopub channel waiting
         in the ZMQ queue.
         """
         idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
         while msg is not None:
             if self.debug:
                 pprint(msg)
             parent = msg['parent_header']
             if not parent or parent['session'] != self.session.session:
                 # ignore IOPub messages not from here
                 idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
                 continue
             msg_id = parent['msg_id']
             content = msg['content']
             header = msg['header']
             msg_type = msg['header']['msg_type']
             
             if msg_type == 'status' and msg_id not in self.metadata:
                 # ignore status messages if they aren't mine
                 idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
                 continue
 
             # init metadata:
             md = self.metadata[msg_id]
 
             if msg_type == 'stream':
                 name = content['name']
                 s = md[name] or ''
                 md[name] = s + content['text']
             elif msg_type == 'error':
                 md.update({'error' : self._unwrap_exception(content)})
             elif msg_type == 'execute_input':
                 md.update({'execute_input' : content['code']})
             elif msg_type == 'display_data':
                 md['outputs'].append(content)
             elif msg_type == 'execute_result':
                 md['execute_result'] = content
             elif msg_type == 'data_message':
                 data, remainder = serialize.deserialize_object(msg['buffers'])
                 md['data'].update(data)
             elif msg_type == 'status':
                 # idle message comes after all outputs
                 if content['execution_state'] == 'idle':
                     md['outputs_ready'] = True
             else:
                 # unhandled msg_type (status, etc.)
                 pass
 
             # reduntant?
             self.metadata[msg_id] = md
 
             idents,msg = self.session.recv(sock, mode=zmq.NOBLOCK)
 
     #--------------------------------------------------------------------------
     # len, getitem
     #--------------------------------------------------------------------------
 
     def __len__(self):
         """len(client) returns # of engines."""
         return len(self.ids)
 
     def __getitem__(self, key):
         """index access returns DirectView multiplexer objects
 
         Must be int, slice, or list/tuple/xrange of ints"""
         if not isinstance(key, (int, slice, tuple, list, xrange)):
             raise TypeError("key by int/slice/iterable of ints only, not %s"%(type(key)))
         else:
             return self.direct_view(key)
     
     def __iter__(self):
         """Since we define getitem, Client is iterable
         
         but unless we also define __iter__, it won't work correctly unless engine IDs
         start at zero and are continuous.
         """
         for eid in self.ids:
             yield self.direct_view(eid)
     
     #--------------------------------------------------------------------------
     # Begin public methods
     #--------------------------------------------------------------------------
 
     @property
     def ids(self):
         """Always up-to-date ids property."""
         self._flush_notifications()
         # always copy:
         return list(self._ids)
 
     def activate(self, targets='all', suffix=''):
         """Create a DirectView and register it with IPython magics
         
         Defines the magics `%px, %autopx, %pxresult, %%px`
         
         Parameters
         ----------
         
         targets: int, list of ints, or 'all'
             The engines on which the view's magics will run
         suffix: str [default: '']
             The suffix, if any, for the magics.  This allows you to have
             multiple views associated with parallel magics at the same time.
             
             e.g. ``rc.activate(targets=0, suffix='0')`` will give you
             the magics ``%px0``, ``%pxresult0``, etc. for running magics just
             on engine 0.
         """
         view = self.direct_view(targets)
         view.block = True
         view.activate(suffix)
         return view
 
     def close(self, linger=None):
         """Close my zmq Sockets
         
         If `linger`, set the zmq LINGER socket option,
         which allows discarding of messages.
         """
         if self._closed:
             return
         self.stop_spin_thread()
         snames = [ trait for trait in self.trait_names() if trait.endswith("socket") ]
         for name in snames:
             socket = getattr(self, name)
             if socket is not None and not socket.closed:
                 if linger is not None:
                     socket.close(linger=linger)
                 else:
                     socket.close()
         self._closed = True
 
     def _spin_every(self, interval=1):
         """target func for use in spin_thread"""
         while True:
             if self._stop_spinning.is_set():
                 return
             time.sleep(interval)
             self.spin()
 
     def spin_thread(self, interval=1):
         """call Client.spin() in a background thread on some regular interval
         
         This helps ensure that messages don't pile up too much in the zmq queue
         while you are working on other things, or just leaving an idle terminal.
         
         It also helps limit potential padding of the `received` timestamp
         on AsyncResult objects, used for timings.
         
         Parameters
         ----------
         
         interval : float, optional
             The interval on which to spin the client in the background thread
             (simply passed to time.sleep).
         
         Notes
         -----
         
         For precision timing, you may want to use this method to put a bound
         on the jitter (in seconds) in `received` timestamps used
         in AsyncResult.wall_time.
         
         """
         if self._spin_thread is not None:
             self.stop_spin_thread()
         self._stop_spinning.clear()
         self._spin_thread = Thread(target=self._spin_every, args=(interval,))
         self._spin_thread.daemon = True
         self._spin_thread.start()
     
     def stop_spin_thread(self):
         """stop background spin_thread, if any"""
         if self._spin_thread is not None:
             self._stop_spinning.set()
             self._spin_thread.join()
             self._spin_thread = None
 
     def spin(self):
         """Flush any registration notifications and execution results
         waiting in the ZMQ queue.
         """
         if self._notification_socket:
             self._flush_notifications()
         if self._iopub_socket:
             self._flush_iopub(self._iopub_socket)
         if self._mux_socket:
             self._flush_results(self._mux_socket)
         if self._task_socket:
             self._flush_results(self._task_socket)
         if self._control_socket:
             self._flush_control(self._control_socket)
         if self._query_socket:
             self._flush_ignored_hub_replies()
 
     def wait(self, jobs=None, timeout=-1):
         """waits on one or more `jobs`, for up to `timeout` seconds.
 
         Parameters
         ----------
 
         jobs : int, str, or list of ints and/or strs, or one or more AsyncResult objects
                 ints are indices to self.history
                 strs are msg_ids
                 default: wait on all outstanding messages
         timeout : float
                 a time in seconds, after which to give up.
                 default is -1, which means no timeout
 
         Returns
         -------
 
         True : when all msg_ids are done
         False : timeout reached, some msg_ids still outstanding
         """
         tic = time.time()
         if jobs is None:
             theids = self.outstanding
         else:
             if isinstance(jobs, string_types + (int, AsyncResult)):
                 jobs = [jobs]
             theids = set()
             for job in jobs:
                 if isinstance(job, int):
                     # index access
                     job = self.history[job]
                 elif isinstance(job, AsyncResult):
                     theids.update(job.msg_ids)
                     continue
                 theids.add(job)
         if not theids.intersection(self.outstanding):
             return True
         self.spin()
         while theids.intersection(self.outstanding):
             if timeout >= 0 and ( time.time()-tic ) > timeout:
                 break
             time.sleep(1e-3)
             self.spin()
         return len(theids.intersection(self.outstanding)) == 0
 
     #--------------------------------------------------------------------------
     # Control methods
     #--------------------------------------------------------------------------
 
     @spin_first
     def clear(self, targets=None, block=None):
         """Clear the namespace in target(s)."""
         block = self.block if block is None else block
         targets = self._build_targets(targets)[0]
         for t in targets:
             self.session.send(self._control_socket, 'clear_request', content={}, ident=t)
         error = False
         if block:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket,0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         if error:
             raise error
 
 
     @spin_first
     def abort(self, jobs=None, targets=None, block=None):
         """Abort specific jobs from the execution queues of target(s).
 
         This is a mechanism to prevent jobs that have already been submitted
         from executing.
 
         Parameters
         ----------
 
         jobs : msg_id, list of msg_ids, or AsyncResult
             The jobs to be aborted
             
             If unspecified/None: abort all outstanding jobs.
 
         """
         block = self.block if block is None else block
         jobs = jobs if jobs is not None else list(self.outstanding)
         targets = self._build_targets(targets)[0]
         
         msg_ids = []
         if isinstance(jobs, string_types + (AsyncResult,)):
             jobs = [jobs]
         bad_ids = [obj for obj in jobs if not isinstance(obj, string_types + (AsyncResult,))]
         if bad_ids:
             raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0])
         for j in jobs:
             if isinstance(j, AsyncResult):
                 msg_ids.extend(j.msg_ids)
             else:
                 msg_ids.append(j)
         content = dict(msg_ids=msg_ids)
         for t in targets:
             self.session.send(self._control_socket, 'abort_request',
                     content=content, ident=t)
         error = False
         if block:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket,0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
         if error:
             raise error
 
     @spin_first
     def shutdown(self, targets='all', restart=False, hub=False, block=None):
         """Terminates one or more engine processes, optionally including the hub.
         
         Parameters
         ----------
         
         targets: list of ints or 'all' [default: all]
             Which engines to shutdown.
         hub: bool [default: False]
             Whether to include the Hub.  hub=True implies targets='all'.
         block: bool [default: self.block]
             Whether to wait for clean shutdown replies or not.
         restart: bool [default: False]
             NOT IMPLEMENTED
             whether to restart engines after shutting them down.
         """
         from IPython.parallel.error import NoEnginesRegistered
         if restart:
             raise NotImplementedError("Engine restart is not yet implemented")
         
         block = self.block if block is None else block
         if hub:
             targets = 'all'
         try:
             targets = self._build_targets(targets)[0]
         except NoEnginesRegistered:
             targets = []
         for t in targets:
             self.session.send(self._control_socket, 'shutdown_request',
                         content={'restart':restart},ident=t)
         error = False
         if block or hub:
             self._flush_ignored_control()
             for i in range(len(targets)):
                 idents,msg = self.session.recv(self._control_socket, 0)
                 if self.debug:
                     pprint(msg)
                 if msg['content']['status'] != 'ok':
                     error = self._unwrap_exception(msg['content'])
         else:
             self._ignored_control_replies += len(targets)
 
         if hub:
             time.sleep(0.25)
             self.session.send(self._query_socket, 'shutdown_request')
             idents,msg = self.session.recv(self._query_socket, 0)
             if self.debug:
                 pprint(msg)
             if msg['content']['status'] != 'ok':
                 error = self._unwrap_exception(msg['content'])
 
         if error:
             raise error
 
     #--------------------------------------------------------------------------
     # Execution related methods
     #--------------------------------------------------------------------------
 
     def _maybe_raise(self, result):
         """wrapper for maybe raising an exception if apply failed."""
         if isinstance(result, error.RemoteError):
             raise result
 
         return result
 
     def send_apply_request(self, socket, f, args=None, kwargs=None, metadata=None, track=False,
                             ident=None):
         """construct and send an apply message via a socket.
 
         This is the principal method with which all engine execution is performed by views.
         """
 
         if self._closed:
             raise RuntimeError("Client cannot be used after its sockets have been closed")
         
         # defaults:
         args = args if args is not None else []
         kwargs = kwargs if kwargs is not None else {}
         metadata = metadata if metadata is not None else {}
 
         # validate arguments
         if not callable(f) and not isinstance(f, Reference):
             raise TypeError("f must be callable, not %s"%type(f))
         if not isinstance(args, (tuple, list)):
             raise TypeError("args must be tuple or list, not %s"%type(args))
         if not isinstance(kwargs, dict):
             raise TypeError("kwargs must be dict, not %s"%type(kwargs))
         if not isinstance(metadata, dict):
             raise TypeError("metadata must be dict, not %s"%type(metadata))
 
         bufs = serialize.pack_apply_message(f, args, kwargs,
             buffer_threshold=self.session.buffer_threshold,
             item_threshold=self.session.item_threshold,
         )
 
         msg = self.session.send(socket, "apply_request", buffers=bufs, ident=ident,
                             metadata=metadata, track=track)
 
         msg_id = msg['header']['msg_id']
         self.outstanding.add(msg_id)
         if ident:
             # possibly routed to a specific engine
             if isinstance(ident, list):
                 ident = ident[-1]
             if ident in self._engines.values():
                 # save for later, in case of engine death
                 self._outstanding_dict[ident].add(msg_id)
         self.history.append(msg_id)
         self.metadata[msg_id]['submitted'] = datetime.now()
 
         return msg
 
     def send_execute_request(self, socket, code, silent=True, metadata=None, ident=None):
         """construct and send an execute request via a socket.
 
         """
 
         if self._closed:
             raise RuntimeError("Client cannot be used after its sockets have been closed")
         
         # defaults:
         metadata = metadata if metadata is not None else {}
 
         # validate arguments
         if not isinstance(code, string_types):
             raise TypeError("code must be text, not %s" % type(code))
         if not isinstance(metadata, dict):
             raise TypeError("metadata must be dict, not %s" % type(metadata))
         
         content = dict(code=code, silent=bool(silent), user_expressions={})
 
 
         msg = self.session.send(socket, "execute_request", content=content, ident=ident,
                             metadata=metadata)
 
         msg_id = msg['header']['msg_id']
         self.outstanding.add(msg_id)
         if ident:
             # possibly routed to a specific engine
             if isinstance(ident, list):
                 ident = ident[-1]
             if ident in self._engines.values():
                 # save for later, in case of engine death
                 self._outstanding_dict[ident].add(msg_id)
         self.history.append(msg_id)
         self.metadata[msg_id]['submitted'] = datetime.now()
 
         return msg
 
     #--------------------------------------------------------------------------
     # construct a View object
     #--------------------------------------------------------------------------
 
     def load_balanced_view(self, targets=None):
         """construct a DirectView object.
 
         If no arguments are specified, create a LoadBalancedView
         using all engines.
 
         Parameters
         ----------
 
         targets: list,slice,int,etc. [default: use all engines]
             The subset of engines across which to load-balance
         """
         if targets == 'all':
             targets = None
         if targets is not None:
             targets = self._build_targets(targets)[1]
         return LoadBalancedView(client=self, socket=self._task_socket, targets=targets)
 
     def direct_view(self, targets='all'):
         """construct a DirectView object.
 
         If no targets are specified, create a DirectView using all engines.
         
         rc.direct_view('all') is distinguished from rc[:] in that 'all' will
         evaluate the target engines at each execution, whereas rc[:] will connect to
         all *current* engines, and that list will not change.
         
         That is, 'all' will always use all engines, whereas rc[:] will not use
         engines added after the DirectView is constructed.
 
         Parameters
         ----------
 
         targets: list,slice,int,etc. [default: use all engines]
             The engines to use for the View
         """
         single = isinstance(targets, int)
         # allow 'all' to be lazily evaluated at each execution
         if targets != 'all':
             targets = self._build_targets(targets)[1]
         if single:
             targets = targets[0]
         return DirectView(client=self, socket=self._mux_socket, targets=targets)
 
     #--------------------------------------------------------------------------
     # Query methods
     #--------------------------------------------------------------------------
 
     @spin_first
     def get_result(self, indices_or_msg_ids=None, block=None, owner=True):
         """Retrieve a result by msg_id or history index, wrapped in an AsyncResult object.
 
         If the client already has the results, no request to the Hub will be made.
 
         This is a convenient way to construct AsyncResult objects, which are wrappers
         that include metadata about execution, and allow for awaiting results that
         were not submitted by this Client.
 
         It can also be a convenient way to retrieve the metadata associated with
         blocking execution, since it always retrieves
 
         Examples
         --------
         ::
 
             In [10]: r = client.apply()
 
         Parameters
         ----------
 
         indices_or_msg_ids : integer history index, str msg_id, or list of either
             The indices or msg_ids of indices to be retrieved
 
         block : bool
             Whether to wait for the result to be done
         owner : bool [default: True]
             Whether this AsyncResult should own the result.
             If so, calling `ar.get()` will remove data from the
             client's result and metadata cache.
             There should only be one owner of any given msg_id.
 
         Returns
         -------
 
         AsyncResult
             A single AsyncResult object will always be returned.
 
         AsyncHubResult
             A subclass of AsyncResult that retrieves results from the Hub
 
         """
         block = self.block if block is None else block
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
         
         single_result = False
         if not isinstance(indices_or_msg_ids, (list,tuple)):
             indices_or_msg_ids = [indices_or_msg_ids]
             single_result = True
 
         theids = []
         for id in indices_or_msg_ids:
             if isinstance(id, int):
                 id = self.history[id]
             if not isinstance(id, string_types):
                 raise TypeError("indices must be str or int, not %r"%id)
             theids.append(id)
 
         local_ids = [msg_id for msg_id in theids if (msg_id in self.outstanding or msg_id in self.results)]
         remote_ids = [msg_id for msg_id in theids if msg_id not in local_ids]
         
         # given single msg_id initially, get_result shot get the result itself,
         # not a length-one list
         if single_result:
             theids = theids[0]
 
         if remote_ids:
             ar = AsyncHubResult(self, msg_ids=theids, owner=owner)
         else:
             ar = AsyncResult(self, msg_ids=theids, owner=owner)
 
         if block:
             ar.wait()
 
         return ar
 
     @spin_first
     def resubmit(self, indices_or_msg_ids=None, metadata=None, block=None):
         """Resubmit one or more tasks.
 
         in-flight tasks may not be resubmitted.
 
         Parameters
         ----------
 
         indices_or_msg_ids : integer history index, str msg_id, or list of either
             The indices or msg_ids of indices to be retrieved
 
         block : bool
             Whether to wait for the result to be done
 
         Returns
         -------
 
         AsyncHubResult
             A subclass of AsyncResult that retrieves results from the Hub
 
         """
         block = self.block if block is None else block
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
 
         if not isinstance(indices_or_msg_ids, (list,tuple)):
             indices_or_msg_ids = [indices_or_msg_ids]
 
         theids = []
         for id in indices_or_msg_ids:
             if isinstance(id, int):
                 id = self.history[id]
             if not isinstance(id, string_types):
                 raise TypeError("indices must be str or int, not %r"%id)
             theids.append(id)
 
         content = dict(msg_ids = theids)
 
         self.session.send(self._query_socket, 'resubmit_request', content)
 
         zmq.select([self._query_socket], [], [])
         idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
         mapping = content['resubmitted']
         new_ids = [ mapping[msg_id] for msg_id in theids ]
 
         ar = AsyncHubResult(self, msg_ids=new_ids)
 
         if block:
             ar.wait()
 
         return ar
 
     @spin_first
     def result_status(self, msg_ids, status_only=True):
         """Check on the status of the result(s) of the apply request with `msg_ids`.
 
         If status_only is False, then the actual results will be retrieved, else
         only the status of the results will be checked.
 
         Parameters
         ----------
 
         msg_ids : list of msg_ids
             if int:
                 Passed as index to self.history for convenience.
         status_only : bool (default: True)
             if False:
                 Retrieve the actual results of completed tasks.
 
         Returns
         -------
 
         results : dict
             There will always be the keys 'pending' and 'completed', which will
             be lists of msg_ids that are incomplete or complete. If `status_only`
             is False, then completed results will be keyed by their `msg_id`.
         """
         if not isinstance(msg_ids, (list,tuple)):
             msg_ids = [msg_ids]
 
         theids = []
         for msg_id in msg_ids:
             if isinstance(msg_id, int):
                 msg_id = self.history[msg_id]
             if not isinstance(msg_id, string_types):
                 raise TypeError("msg_ids must be str, not %r"%msg_id)
             theids.append(msg_id)
 
         completed = []
         local_results = {}
 
         # comment this block out to temporarily disable local shortcut:
         for msg_id in theids:
             if msg_id in self.results:
                 completed.append(msg_id)
                 local_results[msg_id] = self.results[msg_id]
                 theids.remove(msg_id)
 
         if theids: # some not locally cached
             content = dict(msg_ids=theids, status_only=status_only)
             msg = self.session.send(self._query_socket, "result_request", content=content)
             zmq.select([self._query_socket], [], [])
             idents,msg = self.session.recv(self._query_socket, zmq.NOBLOCK)
             if self.debug:
                 pprint(msg)
             content = msg['content']
             if content['status'] != 'ok':
                 raise self._unwrap_exception(content)
             buffers = msg['buffers']
         else:
             content = dict(completed=[],pending=[])
 
         content['completed'].extend(completed)
 
         if status_only:
             return content
 
         failures = []
         # load cached results into result:
         content.update(local_results)
 
         # update cache with results:
         for msg_id in sorted(theids):
             if msg_id in content['completed']:
                 rec = content[msg_id]
                 parent = extract_dates(rec['header'])
                 header = extract_dates(rec['result_header'])
                 rcontent = rec['result_content']
                 iodict = rec['io']
                 if isinstance(rcontent, str):
                     rcontent = self.session.unpack(rcontent)
 
                 md = self.metadata[msg_id]
                 md_msg = dict(
                     content=rcontent,
                     parent_header=parent,
                     header=header,
                     metadata=rec['result_metadata'],
                 )
                 md.update(self._extract_metadata(md_msg))
                 if rec.get('received'):
                     md['received'] = parse_date(rec['received'])
                 md.update(iodict)
                 
                 if rcontent['status'] == 'ok':
                     if header['msg_type'] == 'apply_reply':
                         res,buffers = serialize.deserialize_object(buffers)
                     elif header['msg_type'] == 'execute_reply':
                         res = ExecuteReply(msg_id, rcontent, md)
                     else:
                         raise KeyError("unhandled msg type: %r" % header['msg_type'])
                 else:
                     res = self._unwrap_exception(rcontent)
                     failures.append(res)
 
                 self.results[msg_id] = res
                 content[msg_id] = res
 
         if len(theids) == 1 and failures:
             raise failures[0]
 
         error.collect_exceptions(failures, "result_status")
         return content
 
     @spin_first
     def queue_status(self, targets='all', verbose=False):
         """Fetch the status of engine queues.
 
         Parameters
         ----------
 
         targets : int/str/list of ints/strs
                 the engines whose states are to be queried.
                 default : all
         verbose : bool
                 Whether to return lengths only, or lists of ids for each element
         """
         if targets == 'all':
             # allow 'all' to be evaluated on the engine
             engine_ids = None
         else:
             engine_ids = self._build_targets(targets)[1]
         content = dict(targets=engine_ids, verbose=verbose)
         self.session.send(self._query_socket, "queue_request", content=content)
         idents,msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         status = content.pop('status')
         if status != 'ok':
             raise self._unwrap_exception(content)
         content = rekey(content)
         if isinstance(targets, int):
             return content[targets]
         else:
             return content
 
     def _build_msgids_from_target(self, targets=None):
         """Build a list of msg_ids from the list of engine targets"""
         if not targets: # needed as _build_targets otherwise uses all engines
             return []
         target_ids = self._build_targets(targets)[0] 
         return [md_id for md_id in self.metadata if self.metadata[md_id]["engine_uuid"] in target_ids]
     
     def _build_msgids_from_jobs(self, jobs=None):
         """Build a list of msg_ids from "jobs" """
         if not jobs:
             return []
         msg_ids = []
         if isinstance(jobs, string_types + (AsyncResult,)):
             jobs = [jobs]
         bad_ids = [obj for obj in jobs if not isinstance(obj, string_types + (AsyncResult,))]
         if bad_ids:
             raise TypeError("Invalid msg_id type %r, expected str or AsyncResult"%bad_ids[0])
         for j in jobs:
             if isinstance(j, AsyncResult):
                 msg_ids.extend(j.msg_ids)
             else:
                 msg_ids.append(j)
         return msg_ids        
         
     def purge_local_results(self, jobs=[], targets=[]):
         """Clears the client caches of results and their metadata.
 
         Individual results can be purged by msg_id, or the entire
         history of specific targets can be purged.
 
         Use `purge_local_results('all')` to scrub everything from the Clients's
         results and metadata caches.
 
         After this call all `AsyncResults` are invalid and should be discarded.
 
         If you must "reget" the results, you can still do so by using
         `client.get_result(msg_id)` or `client.get_result(asyncresult)`. This will
         redownload the results from the hub if they are still available
         (i.e `client.purge_hub_results(...)` has not been called.        
 
         Parameters
         ----------
 
         jobs : str or list of str or AsyncResult objects
                 the msg_ids whose results should be purged.
         targets : int/list of ints
                 The engines, by integer ID, whose entire result histories are to be purged.
 
         Raises
         ------
 
         RuntimeError : if any of the tasks to be purged are still outstanding.
 
         """
         if not targets and not jobs:
             raise ValueError("Must specify at least one of `targets` and `jobs`")
         
         if jobs == 'all':
             if self.outstanding:
                 raise RuntimeError("Can't purge outstanding tasks: %s" % self.outstanding)
             self.results.clear()
             self.metadata.clear()
         else:
             msg_ids = set()
             msg_ids.update(self._build_msgids_from_target(targets))
             msg_ids.update(self._build_msgids_from_jobs(jobs))
             still_outstanding = self.outstanding.intersection(msg_ids)
             if still_outstanding:
                 raise RuntimeError("Can't purge outstanding tasks: %s" % still_outstanding)
             for mid in msg_ids:
                 self.results.pop(mid, None)
                 self.metadata.pop(mid, None)
 
 
     @spin_first
     def purge_hub_results(self, jobs=[], targets=[]):
         """Tell the Hub to forget results.
 
         Individual results can be purged by msg_id, or the entire
         history of specific targets can be purged.
 
         Use `purge_results('all')` to scrub everything from the Hub's db.
 
         Parameters
         ----------
 
         jobs : str or list of str or AsyncResult objects
                 the msg_ids whose results should be forgotten.
         targets : int/str/list of ints/strs
                 The targets, by int_id, whose entire history is to be purged.
 
                 default : None
         """
         if not targets and not jobs:
             raise ValueError("Must specify at least one of `targets` and `jobs`")
         if targets:
             targets = self._build_targets(targets)[1]
 
         # construct msg_ids from jobs
         if jobs == 'all':
             msg_ids = jobs
         else:
             msg_ids = self._build_msgids_from_jobs(jobs)
 
         content = dict(engine_ids=targets, msg_ids=msg_ids)
         self.session.send(self._query_socket, "purge_request", content=content)
         idents, msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
 
     def purge_results(self,  jobs=[], targets=[]):
         """Clears the cached results from both the hub and the local client
                 
         Individual results can be purged by msg_id, or the entire
         history of specific targets can be purged.
 
         Use `purge_results('all')` to scrub every cached result from both the Hub's and 
         the Client's db.
         
         Equivalent to calling both `purge_hub_results()` and `purge_client_results()` with 
         the same arguments.
 
         Parameters
         ----------
 
         jobs : str or list of str or AsyncResult objects
                 the msg_ids whose results should be forgotten.
         targets : int/str/list of ints/strs
                 The targets, by int_id, whose entire history is to be purged.
 
                 default : None
         """
         self.purge_local_results(jobs=jobs, targets=targets)
         self.purge_hub_results(jobs=jobs, targets=targets)
 
     def purge_everything(self):
         """Clears all content from previous Tasks from both the hub and the local client
         
         In addition to calling `purge_results("all")` it also deletes the history and 
         other bookkeeping lists.        
         """
         self.purge_results("all")
         self.history = []
         self.session.digest_history.clear()
 
     @spin_first
     def hub_history(self):
         """Get the Hub's history
 
         Just like the Client, the Hub has a history, which is a list of msg_ids.
         This will contain the history of all clients, and, depending on configuration,
         may contain history across multiple cluster sessions.
 
         Any msg_id returned here is a valid argument to `get_result`.
 
         Returns
         -------
 
         msg_ids : list of strs
                 list of all msg_ids, ordered by task submission time.
         """
 
         self.session.send(self._query_socket, "history_request", content={})
         idents, msg = self.session.recv(self._query_socket, 0)
 
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
         else:
             return content['history']
 
     @spin_first
     def db_query(self, query, keys=None):
         """Query the Hub's TaskRecord database
 
         This will return a list of task record dicts that match `query`
 
         Parameters
         ----------
 
         query : mongodb query dict
             The search dict. See mongodb query docs for details.
         keys : list of strs [optional]
             The subset of keys to be returned.  The default is to fetch everything but buffers.
             'msg_id' will *always* be included.
         """
         if isinstance(keys, string_types):
             keys = [keys]
         content = dict(query=query, keys=keys)
         self.session.send(self._query_socket, "db_request", content=content)
         idents, msg = self.session.recv(self._query_socket, 0)
         if self.debug:
             pprint(msg)
         content = msg['content']
         if content['status'] != 'ok':
             raise self._unwrap_exception(content)
 
         records = content['records']
 
         buffer_lens = content['buffer_lens']
         result_buffer_lens = content['result_buffer_lens']
         buffers = msg['buffers']
         has_bufs = buffer_lens is not None
         has_rbufs = result_buffer_lens is not None
         for i,rec in enumerate(records):
             # unpack datetime objects
             for hkey in ('header', 'result_header'):
                 if hkey in rec:
                     rec[hkey] = extract_dates(rec[hkey])
             for dtkey in ('submitted', 'started', 'completed', 'received'):
                 if dtkey in rec:
                     rec[dtkey] = parse_date(rec[dtkey])
             # relink buffers
             if has_bufs:
                 blen = buffer_lens[i]
                 rec['buffers'], buffers = buffers[:blen],buffers[blen:]
             if has_rbufs:
                 blen = result_buffer_lens[i]
                 rec['result_buffers'], buffers = buffers[:blen],buffers[blen:]
 
         return records
 
 __all__ = [ 'Client' ]

IPython/parallel/client/remotefunction.py

@@ -1,276 +1,276 b''
 """Remote Functions and decorators for Views."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import division
 
 import sys
 import warnings
 
-from IPython.external.decorator import decorator
+from decorator import decorator
 from IPython.testing.skipdoctest import skip_doctest
 
 from . import map as Map
 from .asyncresult import AsyncMapResult
 
 #-----------------------------------------------------------------------------
 # Functions and Decorators
 #-----------------------------------------------------------------------------
 
 @skip_doctest
 def remote(view, block=None, **flags):
     """Turn a function into a remote function.
 
     This method can be used for map:
 
     In [1]: @remote(view,block=True)
        ...: def func(a):
        ...:    pass
     """
 
     def remote_function(f):
         return RemoteFunction(view, f, block=block, **flags)
     return remote_function
 
 @skip_doctest
 def parallel(view, dist='b', block=None, ordered=True, **flags):
     """Turn a function into a parallel remote function.
 
     This method can be used for map:
 
     In [1]: @parallel(view, block=True)
        ...: def func(a):
        ...:    pass
     """
 
     def parallel_function(f):
         return ParallelFunction(view, f, dist=dist, block=block, ordered=ordered, **flags)
     return parallel_function
 
 def getname(f):
     """Get the name of an object.
     
     For use in case of callables that are not functions, and
     thus may not have __name__ defined.
     
     Order: f.__name__ >  f.name > str(f)
     """
     try:
         return f.__name__
     except:
         pass
     try:
         return f.name
     except:
         pass
     
     return str(f)
 
 @decorator
 def sync_view_results(f, self, *args, **kwargs):
     """sync relevant results from self.client to our results attribute.
     
     This is a clone of view.sync_results, but for remote functions
     """
     view = self.view
     if view._in_sync_results:
         return f(self, *args, **kwargs)
     view._in_sync_results = True
     try:
         ret = f(self, *args, **kwargs)
     finally:
         view._in_sync_results = False
         view._sync_results()
     return ret
     
 #--------------------------------------------------------------------------
 # Classes
 #--------------------------------------------------------------------------
 
 class RemoteFunction(object):
     """Turn an existing function into a remote function.
 
     Parameters
     ----------
 
     view : View instance
         The view to be used for execution
     f : callable
         The function to be wrapped into a remote function
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `view`
 
     **flags : remaining kwargs are passed to View.temp_flags
     """
 
     view = None # the remote connection
     func = None # the wrapped function
     block = None # whether to block
     flags = None # dict of extra kwargs for temp_flags
 
     def __init__(self, view, f, block=None, **flags):
         self.view = view
         self.func = f
         self.block=block
         self.flags=flags
 
     def __call__(self, *args, **kwargs):
         block = self.view.block if self.block is None else self.block
         with self.view.temp_flags(block=block, **self.flags):
             return self.view.apply(self.func, *args, **kwargs)
 
 
 class ParallelFunction(RemoteFunction):
     """Class for mapping a function to sequences.
 
     This will distribute the sequences according the a mapper, and call
     the function on each sub-sequence.  If called via map, then the function
     will be called once on each element, rather that each sub-sequence.
 
     Parameters
     ----------
 
     view : View instance
         The view to be used for execution
     f : callable
         The function to be wrapped into a remote function
     dist : str [default: 'b']
         The key for which mapObject to use to distribute sequences
         options are:
 
         * 'b' : use contiguous chunks in order
         * 'r' : use round-robin striping
 
     block : bool [default: None]
         Whether to wait for results or not.  The default behavior is
         to use the current `block` attribute of `view`
     chunksize : int or None
         The size of chunk to use when breaking up sequences in a load-balanced manner
     ordered : bool [default: True]
         Whether the result should be kept in order. If False,
         results become available as they arrive, regardless of submission order.
     **flags
         remaining kwargs are passed to View.temp_flags
     """
 
     chunksize = None
     ordered = None
     mapObject = None
     _mapping = False
 
     def __init__(self, view, f, dist='b', block=None, chunksize=None, ordered=True, **flags):
         super(ParallelFunction, self).__init__(view, f, block=block, **flags)
         self.chunksize = chunksize
         self.ordered = ordered
 
         mapClass = Map.dists[dist]
         self.mapObject = mapClass()
     
     @sync_view_results
     def __call__(self, *sequences):
         client = self.view.client
         
         lens = []
         maxlen = minlen = -1
         for i, seq in enumerate(sequences):
             try:
                 n = len(seq)
             except Exception:
                 seq = list(seq)
                 if isinstance(sequences, tuple):
                     # can't alter a tuple
                     sequences = list(sequences)
                 sequences[i] = seq
                 n = len(seq)
             if n > maxlen:
                 maxlen = n
             if minlen == -1 or n < minlen:
                 minlen = n
             lens.append(n)
         
         if maxlen == 0:
             # nothing to iterate over
             return []
         
         # check that the length of sequences match
         if not self._mapping and minlen != maxlen:
             msg = 'all sequences must have equal length, but have %s' % lens
             raise ValueError(msg)
         
         balanced = 'Balanced' in self.view.__class__.__name__
         if balanced:
             if self.chunksize:
                 nparts = maxlen // self.chunksize + int(maxlen % self.chunksize > 0)
             else:
                 nparts = maxlen
             targets = [None]*nparts
         else:
             if self.chunksize:
                 warnings.warn("`chunksize` is ignored unless load balancing", UserWarning)
             # multiplexed:
             targets = self.view.targets
             # 'all' is lazily evaluated at execution time, which is now:
             if targets == 'all':
                 targets = client._build_targets(targets)[1]
             elif isinstance(targets, int):
                 # single-engine view, targets must be iterable
                 targets = [targets]
             nparts = len(targets)
 
         msg_ids = []
         for index, t in enumerate(targets):
             args = []
             for seq in sequences:
                 part = self.mapObject.getPartition(seq, index, nparts, maxlen)
                 args.append(part)
 
             if sum([len(arg) for arg in args]) == 0:
                 continue
 
             if self._mapping:
                 if sys.version_info[0] >= 3:
                     f = lambda f, *sequences: list(map(f, *sequences))
                 else:
                     f = map
                 args = [self.func] + args
             else:
                 f=self.func
 
             view = self.view if balanced else client[t]
             with view.temp_flags(block=False, **self.flags):
                 ar = view.apply(f, *args)
 
             msg_ids.extend(ar.msg_ids)
 
         r = AsyncMapResult(self.view.client, msg_ids, self.mapObject,
                             fname=getname(self.func),
                             ordered=self.ordered
                         )
 
         if self.block:
             try:
                 return r.get()
             except KeyboardInterrupt:
                 return r
         else:
             return r
 
     def map(self, *sequences):
         """call a function on each element of one or more sequence(s) remotely.
         This should behave very much like the builtin map, but return an AsyncMapResult
         if self.block is False.
         
         That means it can take generators (will be cast to lists locally),
         and mismatched sequence lengths will be padded with None.
         """
         # set _mapping as a flag for use inside self.__call__
         self._mapping = True
         try:
             ret = self(*sequences)
         finally:
             self._mapping = False
         return ret
 
 __all__ = ['remote', 'parallel', 'RemoteFunction', 'ParallelFunction']

IPython/parallel/client/view.py

@@ -1,1125 +1,1125 b''
 """Views of remote engines."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import imp
 import sys
 import warnings
 from contextlib import contextmanager
 from types import ModuleType
 
 import zmq
 
 from IPython.testing.skipdoctest import skip_doctest
 from IPython.utils import pickleutil
 from IPython.utils.traitlets import (
     HasTraits, Any, Bool, List, Dict, Set, Instance, CFloat, Integer
 )
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 from IPython.parallel import util
 from IPython.parallel.controller.dependency import Dependency, dependent
 from IPython.utils.py3compat import string_types, iteritems, PY3
 
 from . import map as Map
 from .asyncresult import AsyncResult, AsyncMapResult
 from .remotefunction import ParallelFunction, parallel, remote, getname
 
 #-----------------------------------------------------------------------------
 # Decorators
 #-----------------------------------------------------------------------------
 
 @decorator
 def save_ids(f, self, *args, **kwargs):
     """Keep our history and outstanding attributes up to date after a method call."""
     n_previous = len(self.client.history)
     try:
         ret = f(self, *args, **kwargs)
     finally:
         nmsgs = len(self.client.history) - n_previous
         msg_ids = self.client.history[-nmsgs:]
         self.history.extend(msg_ids)
         self.outstanding.update(msg_ids)
     return ret
 
 @decorator
 def sync_results(f, self, *args, **kwargs):
     """sync relevant results from self.client to our results attribute."""
     if self._in_sync_results:
         return f(self, *args, **kwargs)
     self._in_sync_results = True
     try:
         ret = f(self, *args, **kwargs)
     finally:
         self._in_sync_results = False
         self._sync_results()
     return ret
 
 @decorator
 def spin_after(f, self, *args, **kwargs):
     """call spin after the method."""
     ret = f(self, *args, **kwargs)
     self.spin()
     return ret
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 @skip_doctest
 class View(HasTraits):
     """Base View class for more convenint apply(f,*args,**kwargs) syntax via attributes.
 
     Don't use this class, use subclasses.
 
     Methods
     -------
 
     spin
         flushes incoming results and registration state changes
         control methods spin, and requesting `ids` also ensures up to date
 
     wait
         wait on one or more msg_ids
 
     execution methods
         apply
         legacy: execute, run
 
     data movement
         push, pull, scatter, gather
 
     query methods
         get_result, queue_status, purge_results, result_status
 
     control methods
         abort, shutdown
 
     """
     # flags
     block=Bool(False)
     track=Bool(True)
     targets = Any()
 
     history=List()
     outstanding = Set()
     results = Dict()
     client = Instance('IPython.parallel.Client')
 
     _socket = Instance('zmq.Socket')
     _flag_names = List(['targets', 'block', 'track'])
     _in_sync_results = Bool(False)
     _targets = Any()
     _idents = Any()
 
     def __init__(self, client=None, socket=None, **flags):
         super(View, self).__init__(client=client, _socket=socket)
         self.results = client.results
         self.block = client.block
 
         self.set_flags(**flags)
 
         assert not self.__class__ is View, "Don't use base View objects, use subclasses"
 
     def __repr__(self):
         strtargets = str(self.targets)
         if len(strtargets) > 16:
             strtargets = strtargets[:12]+'...]'
         return "<%s %s>"%(self.__class__.__name__, strtargets)
     
     def __len__(self):
         if isinstance(self.targets, list):
             return len(self.targets)
         elif isinstance(self.targets, int):
             return 1
         else:
             return len(self.client)
     
     def set_flags(self, **kwargs):
         """set my attribute flags by keyword.
 
         Views determine behavior with a few attributes (`block`, `track`, etc.).
         These attributes can be set all at once by name with this method.
 
         Parameters
         ----------
 
         block : bool
             whether to wait for results
         track : bool
             whether to create a MessageTracker to allow the user to
             safely edit after arrays and buffers during non-copying
             sends.
         """
         for name, value in iteritems(kwargs):
             if name not in self._flag_names:
                 raise KeyError("Invalid name: %r"%name)
             else:
                 setattr(self, name, value)
 
     @contextmanager
     def temp_flags(self, **kwargs):
         """temporarily set flags, for use in `with` statements.
 
         See set_flags for permanent setting of flags
 
         Examples
         --------
 
         >>> view.track=False
         ...
         >>> with view.temp_flags(track=True):
         ...    ar = view.apply(dostuff, my_big_array)
         ...    ar.tracker.wait() # wait for send to finish
         >>> view.track
         False
 
         """
         # preflight: save flags, and set temporaries
         saved_flags = {}
         for f in self._flag_names:
             saved_flags[f] = getattr(self, f)
         self.set_flags(**kwargs)
         # yield to the with-statement block
         try:
             yield
         finally:
             # postflight: restore saved flags
             self.set_flags(**saved_flags)
 
 
     #----------------------------------------------------------------
     # apply
     #----------------------------------------------------------------
 
     def _sync_results(self):
         """to be called by @sync_results decorator
         
         after submitting any tasks.
         """
         delta = self.outstanding.difference(self.client.outstanding)
         completed = self.outstanding.intersection(delta)
         self.outstanding = self.outstanding.difference(completed)
         
     @sync_results
     @save_ids
     def _really_apply(self, f, args, kwargs, block=None, **options):
         """wrapper for client.send_apply_request"""
         raise NotImplementedError("Implement in subclasses")
 
     def apply(self, f, *args, **kwargs):
         """calls ``f(*args, **kwargs)`` on remote engines, returning the result.
 
         This method sets all apply flags via this View's attributes.
 
         Returns :class:`~IPython.parallel.client.asyncresult.AsyncResult`
         instance if ``self.block`` is False, otherwise the return value of
         ``f(*args, **kwargs)``.
         """
         return self._really_apply(f, args, kwargs)
 
     def apply_async(self, f, *args, **kwargs):
         """calls ``f(*args, **kwargs)`` on remote engines in a nonblocking manner.
 
         Returns :class:`~IPython.parallel.client.asyncresult.AsyncResult` instance.
         """
         return self._really_apply(f, args, kwargs, block=False)
 
     @spin_after
     def apply_sync(self, f, *args, **kwargs):
         """calls ``f(*args, **kwargs)`` on remote engines in a blocking manner,
          returning the result.
         """
         return self._really_apply(f, args, kwargs, block=True)
 
     #----------------------------------------------------------------
     # wrappers for client and control methods
     #----------------------------------------------------------------
     @sync_results
     def spin(self):
         """spin the client, and sync"""
         self.client.spin()
 
     @sync_results
     def wait(self, jobs=None, timeout=-1):
         """waits on one or more `jobs`, for up to `timeout` seconds.
 
         Parameters
         ----------
 
         jobs : int, str, or list of ints and/or strs, or one or more AsyncResult objects
                 ints are indices to self.history
                 strs are msg_ids
                 default: wait on all outstanding messages
         timeout : float
                 a time in seconds, after which to give up.
                 default is -1, which means no timeout
 
         Returns
         -------
 
         True : when all msg_ids are done
         False : timeout reached, some msg_ids still outstanding
         """
         if jobs is None:
             jobs = self.history
         return self.client.wait(jobs, timeout)
 
     def abort(self, jobs=None, targets=None, block=None):
         """Abort jobs on my engines.
 
         Parameters
         ----------
 
         jobs : None, str, list of strs, optional
             if None: abort all jobs.
             else: abort specific msg_id(s).
         """
         block = block if block is not None else self.block
         targets = targets if targets is not None else self.targets
         jobs = jobs if jobs is not None else list(self.outstanding)
         
         return self.client.abort(jobs=jobs, targets=targets, block=block)
 
     def queue_status(self, targets=None, verbose=False):
         """Fetch the Queue status of my engines"""
         targets = targets if targets is not None else self.targets
         return self.client.queue_status(targets=targets, verbose=verbose)
 
     def purge_results(self, jobs=[], targets=[]):
         """Instruct the controller to forget specific results."""
         if targets is None or targets == 'all':
             targets = self.targets
         return self.client.purge_results(jobs=jobs, targets=targets)
 
     def shutdown(self, targets=None, restart=False, hub=False, block=None):
         """Terminates one or more engine processes, optionally including the hub.
         """
         block = self.block if block is None else block
         if targets is None or targets == 'all':
             targets = self.targets
         return self.client.shutdown(targets=targets, restart=restart, hub=hub, block=block)
 
     @spin_after
     def get_result(self, indices_or_msg_ids=None, block=None, owner=True):
         """return one or more results, specified by history index or msg_id.
 
         See :meth:`IPython.parallel.client.client.Client.get_result` for details.
         """
 
         if indices_or_msg_ids is None:
             indices_or_msg_ids = -1
         if isinstance(indices_or_msg_ids, int):
             indices_or_msg_ids = self.history[indices_or_msg_ids]
         elif isinstance(indices_or_msg_ids, (list,tuple,set)):
             indices_or_msg_ids = list(indices_or_msg_ids)
             for i,index in enumerate(indices_or_msg_ids):
                 if isinstance(index, int):
                     indices_or_msg_ids[i] = self.history[index]
         return self.client.get_result(indices_or_msg_ids, block=block, owner=owner)
 
     #-------------------------------------------------------------------
     # Map
     #-------------------------------------------------------------------
 
     @sync_results
     def map(self, f, *sequences, **kwargs):
         """override in subclasses"""
         raise NotImplementedError
 
     def map_async(self, f, *sequences, **kwargs):
         """Parallel version of builtin :func:`python:map`, using this view's engines.
 
         This is equivalent to ``map(...block=False)``.
 
         See `self.map` for details.
         """
         if 'block' in kwargs:
             raise TypeError("map_async doesn't take a `block` keyword argument.")
         kwargs['block'] = False
         return self.map(f,*sequences,**kwargs)
 
     def map_sync(self, f, *sequences, **kwargs):
         """Parallel version of builtin :func:`python:map`, using this view's engines.
 
         This is equivalent to ``map(...block=True)``.
 
         See `self.map` for details.
         """
         if 'block' in kwargs:
             raise TypeError("map_sync doesn't take a `block` keyword argument.")
         kwargs['block'] = True
         return self.map(f,*sequences,**kwargs)
 
     def imap(self, f, *sequences, **kwargs):
         """Parallel version of :func:`itertools.imap`.
 
         See `self.map` for details.
 
         """
 
         return iter(self.map_async(f,*sequences, **kwargs))
 
     #-------------------------------------------------------------------
     # Decorators
     #-------------------------------------------------------------------
 
     def remote(self, block=None, **flags):
         """Decorator for making a RemoteFunction"""
         block = self.block if block is None else block
         return remote(self, block=block, **flags)
 
     def parallel(self, dist='b', block=None, **flags):
         """Decorator for making a ParallelFunction"""
         block = self.block if block is None else block
         return parallel(self, dist=dist, block=block, **flags)
 
 @skip_doctest
 class DirectView(View):
     """Direct Multiplexer View of one or more engines.
 
     These are created via indexed access to a client:
 
     >>> dv_1 = client[1]
     >>> dv_all = client[:]
     >>> dv_even = client[::2]
     >>> dv_some = client[1:3]
 
     This object provides dictionary access to engine namespaces:
 
     # push a=5:
     >>> dv['a'] = 5
     # pull 'foo':
     >>> dv['foo']
 
     """
 
     def __init__(self, client=None, socket=None, targets=None):
         super(DirectView, self).__init__(client=client, socket=socket, targets=targets)
 
     @property
     def importer(self):
         """sync_imports(local=True) as a property.
 
         See sync_imports for details.
 
         """
         return self.sync_imports(True)
 
     @contextmanager
     def sync_imports(self, local=True, quiet=False):
         """Context Manager for performing simultaneous local and remote imports.
 
         'import x as y' will *not* work.  The 'as y' part will simply be ignored.
         
         If `local=True`, then the package will also be imported locally.
         
         If `quiet=True`, no output will be produced when attempting remote 
         imports. 
         
         Note that remote-only (`local=False`) imports have not been implemented.
         
         >>> with view.sync_imports():
         ...    from numpy import recarray
         importing recarray from numpy on engine(s)
 
         """
         from IPython.utils.py3compat import builtin_mod
         local_import = builtin_mod.__import__
         modules = set()
         results = []
         @util.interactive
         def remote_import(name, fromlist, level):
             """the function to be passed to apply, that actually performs the import
             on the engine, and loads up the user namespace.
             """
             import sys
             user_ns = globals()
             mod = __import__(name, fromlist=fromlist, level=level)
             if fromlist:
                 for key in fromlist:
                     user_ns[key] = getattr(mod, key)
             else:
                 user_ns[name] = sys.modules[name]
 
         def view_import(name, globals={}, locals={}, fromlist=[], level=0):
             """the drop-in replacement for __import__, that optionally imports
             locally as well.
             """
             # don't override nested imports
             save_import = builtin_mod.__import__
             builtin_mod.__import__ = local_import
 
             if imp.lock_held():
                 # this is a side-effect import, don't do it remotely, or even
                 # ignore the local effects
                 return local_import(name, globals, locals, fromlist, level)
 
             imp.acquire_lock()
             if local:
                 mod = local_import(name, globals, locals, fromlist, level)
             else:
                 raise NotImplementedError("remote-only imports not yet implemented")
             imp.release_lock()
 
             key = name+':'+','.join(fromlist or [])
             if level <= 0 and key not in modules:
                 modules.add(key)
                 if not quiet:
                     if fromlist:
                         print("importing %s from %s on engine(s)"%(','.join(fromlist), name))
                     else:
                         print("importing %s on engine(s)"%name)
                 results.append(self.apply_async(remote_import, name, fromlist, level))
             # restore override
             builtin_mod.__import__ = save_import
 
             return mod
 
         # override __import__
         builtin_mod.__import__ = view_import
         try:
             # enter the block
             yield
         except ImportError:
             if local:
                 raise
             else:
                 # ignore import errors if not doing local imports
                 pass
         finally:
             # always restore __import__
             builtin_mod.__import__ = local_import
 
         for r in results:
             # raise possible remote ImportErrors here
             r.get()
     
     def use_dill(self):
         """Expand serialization support with dill
         
         adds support for closures, etc.
         
         This calls IPython.utils.pickleutil.use_dill() here and on each engine.
         """
         pickleutil.use_dill()
         return self.apply(pickleutil.use_dill)
 
     def use_cloudpickle(self):
         """Expand serialization support with cloudpickle.
         """
         pickleutil.use_cloudpickle()
         return self.apply(pickleutil.use_cloudpickle)
 
 
     @sync_results
     @save_ids
     def _really_apply(self, f, args=None, kwargs=None, targets=None, block=None, track=None):
         """calls f(*args, **kwargs) on remote engines, returning the result.
 
         This method sets all of `apply`'s flags via this View's attributes.
 
         Parameters
         ----------
 
         f : callable
 
         args : list [default: empty]
 
         kwargs : dict [default: empty]
 
         targets : target list [default: self.targets]
             where to run
         block : bool [default: self.block]
             whether to block
         track : bool [default: self.track]
             whether to ask zmq to track the message, for safe non-copying sends
 
         Returns
         -------
 
         if self.block is False:
             returns AsyncResult
         else:
             returns actual result of f(*args, **kwargs) on the engine(s)
             This will be a list of self.targets is also a list (even length 1), or
             the single result if self.targets is an integer engine id
         """
         args = [] if args is None else args
         kwargs = {} if kwargs is None else kwargs
         block = self.block if block is None else block
         track = self.track if track is None else track
         targets = self.targets if targets is None else targets
         
         _idents, _targets = self.client._build_targets(targets)
         msg_ids = []
         trackers = []
         for ident in _idents:
             msg = self.client.send_apply_request(self._socket, f, args, kwargs, track=track,
                                     ident=ident)
             if track:
                 trackers.append(msg['tracker'])
             msg_ids.append(msg['header']['msg_id'])
         if isinstance(targets, int):
             msg_ids = msg_ids[0]
         tracker = None if track is False else zmq.MessageTracker(*trackers)
         ar = AsyncResult(self.client, msg_ids, fname=getname(f), targets=_targets,
             tracker=tracker, owner=True,
         )
         if block:
             try:
                 return ar.get()
             except KeyboardInterrupt:
                 pass
         return ar
 
 
     @sync_results
     def map(self, f, *sequences, **kwargs):
         """``view.map(f, *sequences, block=self.block)`` => list|AsyncMapResult
 
         Parallel version of builtin `map`, using this View's `targets`.
 
         There will be one task per target, so work will be chunked
         if the sequences are longer than `targets`.
 
         Results can be iterated as they are ready, but will become available in chunks.
 
         Parameters
         ----------
 
         f : callable
             function to be mapped
         *sequences: one or more sequences of matching length
             the sequences to be distributed and passed to `f`
         block : bool
             whether to wait for the result or not [default self.block]
 
         Returns
         -------
 
 
         If block=False
           An :class:`~IPython.parallel.client.asyncresult.AsyncMapResult` instance.
           An object like AsyncResult, but which reassembles the sequence of results
           into a single list. AsyncMapResults can be iterated through before all
           results are complete.
         else
             A list, the result of ``map(f,*sequences)``
         """
 
         block = kwargs.pop('block', self.block)
         for k in kwargs.keys():
             if k not in ['block', 'track']:
                 raise TypeError("invalid keyword arg, %r"%k)
 
         assert len(sequences) > 0, "must have some sequences to map onto!"
         pf = ParallelFunction(self, f, block=block, **kwargs)
         return pf.map(*sequences)
 
     @sync_results
     @save_ids
     def execute(self, code, silent=True, targets=None, block=None):
         """Executes `code` on `targets` in blocking or nonblocking manner.
 
         ``execute`` is always `bound` (affects engine namespace)
 
         Parameters
         ----------
 
         code : str
                 the code string to be executed
         block : bool
                 whether or not to wait until done to return
                 default: self.block
         """
         block = self.block if block is None else block
         targets = self.targets if targets is None else targets
 
         _idents, _targets = self.client._build_targets(targets)
         msg_ids = []
         trackers = []
         for ident in _idents:
             msg = self.client.send_execute_request(self._socket, code, silent=silent, ident=ident)
             msg_ids.append(msg['header']['msg_id'])
         if isinstance(targets, int):
             msg_ids = msg_ids[0]
         ar = AsyncResult(self.client, msg_ids, fname='execute', targets=_targets, owner=True)
         if block:
             try:
                 ar.get()
             except KeyboardInterrupt:
                 pass
         return ar
 
     def run(self, filename, targets=None, block=None):
         """Execute contents of `filename` on my engine(s).
 
         This simply reads the contents of the file and calls `execute`.
 
         Parameters
         ----------
 
         filename : str
                 The path to the file
         targets : int/str/list of ints/strs
                 the engines on which to execute
                 default : all
         block : bool
                 whether or not to wait until done
                 default: self.block
 
         """
         with open(filename, 'r') as f:
             # add newline in case of trailing indented whitespace
             # which will cause SyntaxError
             code = f.read()+'\n'
         return self.execute(code, block=block, targets=targets)
 
     def update(self, ns):
         """update remote namespace with dict `ns`
 
         See `push` for details.
         """
         return self.push(ns, block=self.block, track=self.track)
 
     def push(self, ns, targets=None, block=None, track=None):
         """update remote namespace with dict `ns`
 
         Parameters
         ----------
 
         ns : dict
             dict of keys with which to update engine namespace(s)
         block : bool [default : self.block]
             whether to wait to be notified of engine receipt
 
         """
 
         block = block if block is not None else self.block
         track = track if track is not None else self.track
         targets = targets if targets is not None else self.targets
         # applier = self.apply_sync if block else self.apply_async
         if not isinstance(ns, dict):
             raise TypeError("Must be a dict, not %s"%type(ns))
         return self._really_apply(util._push, kwargs=ns, block=block, track=track, targets=targets)
 
     def get(self, key_s):
         """get object(s) by `key_s` from remote namespace
 
         see `pull` for details.
         """
         # block = block if block is not None else self.block
         return self.pull(key_s, block=True)
 
     def pull(self, names, targets=None, block=None):
         """get object(s) by `name` from remote namespace
 
         will return one object if it is a key.
         can also take a list of keys, in which case it will return a list of objects.
         """
         block = block if block is not None else self.block
         targets = targets if targets is not None else self.targets
         applier = self.apply_sync if block else self.apply_async
         if isinstance(names, string_types):
             pass
         elif isinstance(names, (list,tuple,set)):
             for key in names:
                 if not isinstance(key, string_types):
                     raise TypeError("keys must be str, not type %r"%type(key))
         else:
             raise TypeError("names must be strs, not %r"%names)
         return self._really_apply(util._pull, (names,), block=block, targets=targets)
 
     def scatter(self, key, seq, dist='b', flatten=False, targets=None, block=None, track=None):
         """
         Partition a Python sequence and send the partitions to a set of engines.
         """
         block = block if block is not None else self.block
         track = track if track is not None else self.track
         targets = targets if targets is not None else self.targets
         
         # construct integer ID list:
         targets = self.client._build_targets(targets)[1]
 
         mapObject = Map.dists[dist]()
         nparts = len(targets)
         msg_ids = []
         trackers = []
         for index, engineid in enumerate(targets):
             partition = mapObject.getPartition(seq, index, nparts)
             if flatten and len(partition) == 1:
                 ns = {key: partition[0]}
             else:
                 ns = {key: partition}
             r = self.push(ns, block=False, track=track, targets=engineid)
             msg_ids.extend(r.msg_ids)
             if track:
                 trackers.append(r._tracker)
 
         if track:
             tracker = zmq.MessageTracker(*trackers)
         else:
             tracker = None
 
         r = AsyncResult(self.client, msg_ids, fname='scatter', targets=targets,
             tracker=tracker, owner=True,
         )
         if block:
             r.wait()
         else:
             return r
 
     @sync_results
     @save_ids
     def gather(self, key, dist='b', targets=None, block=None):
         """
         Gather a partitioned sequence on a set of engines as a single local seq.
         """
         block = block if block is not None else self.block
         targets = targets if targets is not None else self.targets
         mapObject = Map.dists[dist]()
         msg_ids = []
 
         # construct integer ID list:
         targets = self.client._build_targets(targets)[1]
         
         for index, engineid in enumerate(targets):
             msg_ids.extend(self.pull(key, block=False, targets=engineid).msg_ids)
 
         r = AsyncMapResult(self.client, msg_ids, mapObject, fname='gather')
 
         if block:
             try:
                 return r.get()
             except KeyboardInterrupt:
                 pass
         return r
 
     def __getitem__(self, key):
         return self.get(key)
 
     def __setitem__(self,key, value):
         self.update({key:value})
 
     def clear(self, targets=None, block=None):
         """Clear the remote namespaces on my engines."""
         block = block if block is not None else self.block
         targets = targets if targets is not None else self.targets
         return self.client.clear(targets=targets, block=block)
 
     #----------------------------------------
     # activate for %px, %autopx, etc. magics
     #----------------------------------------
 
     def activate(self, suffix=''):
         """Activate IPython magics associated with this View
         
         Defines the magics `%px, %autopx, %pxresult, %%px, %pxconfig`
         
         Parameters
         ----------
         
         suffix: str [default: '']
             The suffix, if any, for the magics.  This allows you to have
             multiple views associated with parallel magics at the same time.
             
             e.g. ``rc[::2].activate(suffix='_even')`` will give you
             the magics ``%px_even``, ``%pxresult_even``, etc. for running magics 
             on the even engines.
         """
         
         from IPython.parallel.client.magics import ParallelMagics
         
         try:
             # This is injected into __builtins__.
             ip = get_ipython()
         except NameError:
             print("The IPython parallel magics (%px, etc.) only work within IPython.")
             return
         
         M = ParallelMagics(ip, self, suffix)
         ip.magics_manager.register(M)
 
 
 @skip_doctest
 class LoadBalancedView(View):
     """An load-balancing View that only executes via the Task scheduler.
 
     Load-balanced views can be created with the client's `view` method:
 
     >>> v = client.load_balanced_view()
 
     or targets can be specified, to restrict the potential destinations:
 
     >>> v = client.load_balanced_view([1,3])
 
     which would restrict loadbalancing to between engines 1 and 3.
 
     """
 
     follow=Any()
     after=Any()
     timeout=CFloat()
     retries = Integer(0)
 
     _task_scheme = Any()
     _flag_names = List(['targets', 'block', 'track', 'follow', 'after', 'timeout', 'retries'])
 
     def __init__(self, client=None, socket=None, **flags):
         super(LoadBalancedView, self).__init__(client=client, socket=socket, **flags)
         self._task_scheme=client._task_scheme
 
     def _validate_dependency(self, dep):
         """validate a dependency.
 
         For use in `set_flags`.
         """
         if dep is None or isinstance(dep, string_types + (AsyncResult, Dependency)):
             return True
         elif isinstance(dep, (list,set, tuple)):
             for d in dep:
                 if not isinstance(d, string_types + (AsyncResult,)):
                     return False
         elif isinstance(dep, dict):
             if set(dep.keys()) != set(Dependency().as_dict().keys()):
                 return False
             if not isinstance(dep['msg_ids'], list):
                 return False
             for d in dep['msg_ids']:
                 if not isinstance(d, string_types):
                     return False
         else:
             return False
 
         return True
 
     def _render_dependency(self, dep):
         """helper for building jsonable dependencies from various input forms."""
         if isinstance(dep, Dependency):
             return dep.as_dict()
         elif isinstance(dep, AsyncResult):
             return dep.msg_ids
         elif dep is None:
             return []
         else:
             # pass to Dependency constructor
             return list(Dependency(dep))
 
     def set_flags(self, **kwargs):
         """set my attribute flags by keyword.
 
         A View is a wrapper for the Client's apply method, but with attributes
         that specify keyword arguments, those attributes can be set by keyword
         argument with this method.
 
         Parameters
         ----------
 
         block : bool
             whether to wait for results
         track : bool
             whether to create a MessageTracker to allow the user to
             safely edit after arrays and buffers during non-copying
             sends.
 
         after : Dependency or collection of msg_ids
             Only for load-balanced execution (targets=None)
             Specify a list of msg_ids as a time-based dependency.
             This job will only be run *after* the dependencies
             have been met.
 
         follow : Dependency or collection of msg_ids
             Only for load-balanced execution (targets=None)
             Specify a list of msg_ids as a location-based dependency.
             This job will only be run on an engine where this dependency
             is met.
 
         timeout : float/int or None
             Only for load-balanced execution (targets=None)
             Specify an amount of time (in seconds) for the scheduler to
             wait for dependencies to be met before failing with a
             DependencyTimeout.
 
         retries : int
             Number of times a task will be retried on failure.
         """
 
         super(LoadBalancedView, self).set_flags(**kwargs)
         for name in ('follow', 'after'):
             if name in kwargs:
                 value = kwargs[name]
                 if self._validate_dependency(value):
                     setattr(self, name, value)
                 else:
                     raise ValueError("Invalid dependency: %r"%value)
         if 'timeout' in kwargs:
             t = kwargs['timeout']
             if not isinstance(t, (int, float, type(None))):
                 if (not PY3) and (not isinstance(t, long)):
                     raise TypeError("Invalid type for timeout: %r"%type(t))
             if t is not None:
                 if t < 0:
                     raise ValueError("Invalid timeout: %s"%t)
             self.timeout = t
 
     @sync_results
     @save_ids
     def _really_apply(self, f, args=None, kwargs=None, block=None, track=None,
                                         after=None, follow=None, timeout=None,
                                         targets=None, retries=None):
         """calls f(*args, **kwargs) on a remote engine, returning the result.
 
         This method temporarily sets all of `apply`'s flags for a single call.
 
         Parameters
         ----------
 
         f : callable
 
         args : list [default: empty]
 
         kwargs : dict [default: empty]
 
         block : bool [default: self.block]
             whether to block
         track : bool [default: self.track]
             whether to ask zmq to track the message, for safe non-copying sends
 
         !!!!!! TODO: THE REST HERE  !!!!
 
         Returns
         -------
 
         if self.block is False:
             returns AsyncResult
         else:
             returns actual result of f(*args, **kwargs) on the engine(s)
             This will be a list of self.targets is also a list (even length 1), or
             the single result if self.targets is an integer engine id
         """
 
         # validate whether we can run
         if self._socket.closed:
             msg = "Task farming is disabled"
             if self._task_scheme == 'pure':
                 msg += " because the pure ZMQ scheduler cannot handle"
                 msg += " disappearing engines."
             raise RuntimeError(msg)
 
         if self._task_scheme == 'pure':
             # pure zmq scheme doesn't support extra features
             msg = "Pure ZMQ scheduler doesn't support the following flags:"
             "follow, after, retries, targets, timeout"
             if (follow or after or retries or targets or timeout):
                 # hard fail on Scheduler flags
                 raise RuntimeError(msg)
             if isinstance(f, dependent):
                 # soft warn on functional dependencies
                 warnings.warn(msg, RuntimeWarning)
 
         # build args
         args = [] if args is None else args
         kwargs = {} if kwargs is None else kwargs
         block = self.block if block is None else block
         track = self.track if track is None else track
         after = self.after if after is None else after
         retries = self.retries if retries is None else retries
         follow = self.follow if follow is None else follow
         timeout = self.timeout if timeout is None else timeout
         targets = self.targets if targets is None else targets
 
         if not isinstance(retries, int):
             raise TypeError('retries must be int, not %r'%type(retries))
 
         if targets is None:
             idents = []
         else:
             idents = self.client._build_targets(targets)[0]
             # ensure *not* bytes
             idents = [ ident.decode() for ident in idents ]
 
         after = self._render_dependency(after)
         follow = self._render_dependency(follow)
         metadata = dict(after=after, follow=follow, timeout=timeout, targets=idents, retries=retries)
 
         msg = self.client.send_apply_request(self._socket, f, args, kwargs, track=track,
                                 metadata=metadata)
         tracker = None if track is False else msg['tracker']
 
         ar = AsyncResult(self.client, msg['header']['msg_id'], fname=getname(f),
             targets=None, tracker=tracker, owner=True,
         )
         if block:
             try:
                 return ar.get()
             except KeyboardInterrupt:
                 pass
         return ar
 
     @sync_results
     @save_ids
     def map(self, f, *sequences, **kwargs):
         """``view.map(f, *sequences, block=self.block, chunksize=1, ordered=True)`` => list|AsyncMapResult
 
         Parallel version of builtin `map`, load-balanced by this View.
 
         `block`, and `chunksize` can be specified by keyword only.
 
         Each `chunksize` elements will be a separate task, and will be
         load-balanced. This lets individual elements be available for iteration
         as soon as they arrive.
 
         Parameters
         ----------
 
         f : callable
             function to be mapped
         *sequences: one or more sequences of matching length
             the sequences to be distributed and passed to `f`
         block : bool [default self.block]
             whether to wait for the result or not
         track : bool
             whether to create a MessageTracker to allow the user to
             safely edit after arrays and buffers during non-copying
             sends.
         chunksize : int [default 1]
             how many elements should be in each task.
         ordered : bool [default True]
             Whether the results should be gathered as they arrive, or enforce
             the order of submission.
             
             Only applies when iterating through AsyncMapResult as results arrive.
             Has no effect when block=True.
 
         Returns
         -------
 
         if block=False
           An :class:`~IPython.parallel.client.asyncresult.AsyncMapResult` instance.
           An object like AsyncResult, but which reassembles the sequence of results
           into a single list. AsyncMapResults can be iterated through before all
           results are complete.
         else
             A list, the result of ``map(f,*sequences)``
         """
 
         # default
         block = kwargs.get('block', self.block)
         chunksize = kwargs.get('chunksize', 1)
         ordered = kwargs.get('ordered', True)
 
         keyset = set(kwargs.keys())
         extra_keys = keyset.difference_update(set(['block', 'chunksize']))
         if extra_keys:
             raise TypeError("Invalid kwargs: %s"%list(extra_keys))
 
         assert len(sequences) > 0, "must have some sequences to map onto!"
 
         pf = ParallelFunction(self, f, block=block, chunksize=chunksize, ordered=ordered)
         return pf.map(*sequences)
 
 __all__ = ['LoadBalancedView', 'DirectView']

IPython/parallel/controller/scheduler.py

@@ -1,849 +1,849 b''
 """The Python scheduler for rich scheduling.
 
 The Pure ZMQ scheduler does not allow routing schemes other than LRU,
 nor does it check msg_id DAG dependencies. For those, a slightly slower
 Python Scheduler exists.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import logging
 import sys
 import time
 
 from collections import deque
 from datetime import datetime
 from random import randint, random
 from types import FunctionType
 
 try:
     import numpy
 except ImportError:
     numpy = None
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # local imports
-from IPython.external.decorator import decorator
+from decorator import decorator
 from IPython.config.application import Application
 from IPython.config.loader import Config
 from IPython.utils.traitlets import Instance, Dict, List, Set, Integer, Enum, CBytes
 from IPython.utils.py3compat import cast_bytes
 
 from IPython.parallel import error, util
 from IPython.parallel.factory import SessionFactory
 from IPython.parallel.util import connect_logger, local_logger
 
 from .dependency import Dependency
 
 @decorator
 def logged(f,self,*args,**kwargs):
     # print ("#--------------------")
     self.log.debug("scheduler::%s(*%s,**%s)", f.__name__, args, kwargs)
     # print ("#--")
     return f(self,*args, **kwargs)
 
 #----------------------------------------------------------------------
 # Chooser functions
 #----------------------------------------------------------------------
 
 def plainrandom(loads):
     """Plain random pick."""
     n = len(loads)
     return randint(0,n-1)
 
 def lru(loads):
     """Always pick the front of the line.
 
     The content of `loads` is ignored.
 
     Assumes LRU ordering of loads, with oldest first.
     """
     return 0
 
 def twobin(loads):
     """Pick two at random, use the LRU of the two.
 
     The content of loads is ignored.
 
     Assumes LRU ordering of loads, with oldest first.
     """
     n = len(loads)
     a = randint(0,n-1)
     b = randint(0,n-1)
     return min(a,b)
 
 def weighted(loads):
     """Pick two at random using inverse load as weight.
 
     Return the less loaded of the two.
     """
     # weight 0 a million times more than 1:
     weights = 1./(1e-6+numpy.array(loads))
     sums = weights.cumsum()
     t = sums[-1]
     x = random()*t
     y = random()*t
     idx = 0
     idy = 0
     while sums[idx] < x:
         idx += 1
     while sums[idy] < y:
         idy += 1
     if weights[idy] > weights[idx]:
         return idy
     else:
         return idx
 
 def leastload(loads):
     """Always choose the lowest load.
 
     If the lowest load occurs more than once, the first
     occurance will be used.  If loads has LRU ordering, this means
     the LRU of those with the lowest load is chosen.
     """
     return loads.index(min(loads))
 
 #---------------------------------------------------------------------
 # Classes
 #---------------------------------------------------------------------
 
 
 # store empty default dependency:
 MET = Dependency([])
 
 
 class Job(object):
     """Simple container for a job"""
     def __init__(self, msg_id, raw_msg, idents, msg, header, metadata,
                     targets, after, follow, timeout):
         self.msg_id = msg_id
         self.raw_msg = raw_msg
         self.idents = idents
         self.msg = msg
         self.header = header
         self.metadata = metadata
         self.targets = targets
         self.after = after
         self.follow = follow
         self.timeout = timeout
         
         self.removed = False # used for lazy-delete from sorted queue
         self.timestamp = time.time()
         self.timeout_id = 0
         self.blacklist = set()
 
     def __lt__(self, other):
         return self.timestamp < other.timestamp
 
     def __cmp__(self, other):
         return cmp(self.timestamp, other.timestamp)
 
     @property
     def dependents(self):
         return self.follow.union(self.after)
 
 
 class TaskScheduler(SessionFactory):
     """Python TaskScheduler object.
 
     This is the simplest object that supports msg_id based
     DAG dependencies. *Only* task msg_ids are checked, not
     msg_ids of jobs submitted via the MUX queue.
 
     """
 
     hwm = Integer(1, config=True,
         help="""specify the High Water Mark (HWM) for the downstream
         socket in the Task scheduler. This is the maximum number
         of allowed outstanding tasks on each engine.
         
         The default (1) means that only one task can be outstanding on each
         engine.  Setting TaskScheduler.hwm=0 means there is no limit, and the
         engines continue to be assigned tasks while they are working,
         effectively hiding network latency behind computation, but can result
         in an imbalance of work when submitting many heterogenous tasks all at
         once.  Any positive value greater than one is a compromise between the
         two.
 
         """
     )
     scheme_name = Enum(('leastload', 'pure', 'lru', 'plainrandom', 'weighted', 'twobin'),
         'leastload', config=True,
 help="""select the task scheduler scheme  [default: Python LRU]
         Options are: 'pure', 'lru', 'plainrandom', 'weighted', 'twobin','leastload'"""
     )
     def _scheme_name_changed(self, old, new):
         self.log.debug("Using scheme %r"%new)
         self.scheme = globals()[new]
 
     # input arguments:
     scheme = Instance(FunctionType) # function for determining the destination
     def _scheme_default(self):
         return leastload
     client_stream = Instance(zmqstream.ZMQStream) # client-facing stream
     engine_stream = Instance(zmqstream.ZMQStream) # engine-facing stream
     notifier_stream = Instance(zmqstream.ZMQStream) # hub-facing sub stream
     mon_stream = Instance(zmqstream.ZMQStream) # hub-facing pub stream
     query_stream = Instance(zmqstream.ZMQStream) # hub-facing DEALER stream
 
     # internals:
     queue = Instance(deque) # sorted list of Jobs
     def _queue_default(self):
         return deque()
     queue_map = Dict() # dict by msg_id of Jobs (for O(1) access to the Queue)
     graph = Dict() # dict by msg_id of [ msg_ids that depend on key ]
     retries = Dict() # dict by msg_id of retries remaining (non-neg ints)
     # waiting = List() # list of msg_ids ready to run, but haven't due to HWM
     pending = Dict() # dict by engine_uuid of submitted tasks
     completed = Dict() # dict by engine_uuid of completed tasks
     failed = Dict() # dict by engine_uuid of failed tasks
     destinations = Dict() # dict by msg_id of engine_uuids where jobs ran (reverse of completed+failed)
     clients = Dict() # dict by msg_id for who submitted the task
     targets = List() # list of target IDENTs
     loads = List() # list of engine loads
     # full = Set() # set of IDENTs that have HWM outstanding tasks
     all_completed = Set() # set of all completed tasks
     all_failed = Set() # set of all failed tasks
     all_done = Set() # set of all finished tasks=union(completed,failed)
     all_ids = Set() # set of all submitted task IDs
 
     ident = CBytes() # ZMQ identity. This should just be self.session.session
                      # but ensure Bytes
     def _ident_default(self):
         return self.session.bsession
 
     def start(self):
         self.query_stream.on_recv(self.dispatch_query_reply)
         self.session.send(self.query_stream, "connection_request", {})
         
         self.engine_stream.on_recv(self.dispatch_result, copy=False)
         self.client_stream.on_recv(self.dispatch_submission, copy=False)
 
         self._notification_handlers = dict(
             registration_notification = self._register_engine,
             unregistration_notification = self._unregister_engine
         )
         self.notifier_stream.on_recv(self.dispatch_notification)
         self.log.info("Scheduler started [%s]" % self.scheme_name)
 
     def resume_receiving(self):
         """Resume accepting jobs."""
         self.client_stream.on_recv(self.dispatch_submission, copy=False)
 
     def stop_receiving(self):
         """Stop accepting jobs while there are no engines.
         Leave them in the ZMQ queue."""
         self.client_stream.on_recv(None)
 
     #-----------------------------------------------------------------------
     # [Un]Registration Handling
     #-----------------------------------------------------------------------
     
     
     def dispatch_query_reply(self, msg):
         """handle reply to our initial connection request"""
         try:
             idents,msg = self.session.feed_identities(msg)
         except ValueError:
             self.log.warn("task::Invalid Message: %r",msg)
             return
         try:
             msg = self.session.deserialize(msg)
         except ValueError:
             self.log.warn("task::Unauthorized message from: %r"%idents)
             return
         
         content = msg['content']
         for uuid in content.get('engines', {}).values():
             self._register_engine(cast_bytes(uuid))
 
     
     @util.log_errors
     def dispatch_notification(self, msg):
         """dispatch register/unregister events."""
         try:
             idents,msg = self.session.feed_identities(msg)
         except ValueError:
             self.log.warn("task::Invalid Message: %r",msg)
             return
         try:
             msg = self.session.deserialize(msg)
         except ValueError:
             self.log.warn("task::Unauthorized message from: %r"%idents)
             return
 
         msg_type = msg['header']['msg_type']
 
         handler = self._notification_handlers.get(msg_type, None)
         if handler is None:
             self.log.error("Unhandled message type: %r"%msg_type)
         else:
             try:
                 handler(cast_bytes(msg['content']['uuid']))
             except Exception:
                 self.log.error("task::Invalid notification msg: %r", msg, exc_info=True)
 
     def _register_engine(self, uid):
         """New engine with ident `uid` became available."""
         # head of the line:
         self.targets.insert(0,uid)
         self.loads.insert(0,0)
 
         # initialize sets
         self.completed[uid] = set()
         self.failed[uid] = set()
         self.pending[uid] = {}
 
         # rescan the graph:
         self.update_graph(None)
 
     def _unregister_engine(self, uid):
         """Existing engine with ident `uid` became unavailable."""
         if len(self.targets) == 1:
             # this was our only engine
             pass
 
         # handle any potentially finished tasks:
         self.engine_stream.flush()
 
         # don't pop destinations, because they might be used later
         # map(self.destinations.pop, self.completed.pop(uid))
         # map(self.destinations.pop, self.failed.pop(uid))
 
         # prevent this engine from receiving work
         idx = self.targets.index(uid)
         self.targets.pop(idx)
         self.loads.pop(idx)
 
         # wait 5 seconds before cleaning up pending jobs, since the results might
         # still be incoming
         if self.pending[uid]:
             self.loop.add_timeout(self.loop.time() + 5,
                 lambda : self.handle_stranded_tasks(uid),
             )
         else:
             self.completed.pop(uid)
             self.failed.pop(uid)
 
 
     def handle_stranded_tasks(self, engine):
         """Deal with jobs resident in an engine that died."""
         lost = self.pending[engine]
         for msg_id in lost.keys():
             if msg_id not in self.pending[engine]:
                 # prevent double-handling of messages
                 continue
 
             raw_msg = lost[msg_id].raw_msg
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             parent = self.session.unpack(msg[1].bytes)
             idents = [engine, idents[0]]
 
             # build fake error reply
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(engine, msg_id))
             except:
                 content = error.wrap_exception()
             # build fake metadata
             md = dict(
                 status=u'error',
                 engine=engine.decode('ascii'),
                 date=datetime.now(),
             )
             msg = self.session.msg('apply_reply', content, parent=parent, metadata=md)
             raw_reply = list(map(zmq.Message, self.session.serialize(msg, ident=idents)))
             # and dispatch it
             self.dispatch_result(raw_reply)
 
         # finally scrub completed/failed lists
         self.completed.pop(engine)
         self.failed.pop(engine)
 
 
     #-----------------------------------------------------------------------
     # Job Submission
     #-----------------------------------------------------------------------
     
 
     @util.log_errors
     def dispatch_submission(self, raw_msg):
         """Dispatch job submission to appropriate handlers."""
         # ensure targets up to date:
         self.notifier_stream.flush()
         try:
             idents, msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.deserialize(msg, content=False, copy=False)
         except Exception:
             self.log.error("task::Invaid task msg: %r"%raw_msg, exc_info=True)
             return
 
 
         # send to monitor
         self.mon_stream.send_multipart([b'intask']+raw_msg, copy=False)
 
         header = msg['header']
         md = msg['metadata']
         msg_id = header['msg_id']
         self.all_ids.add(msg_id)
 
         # get targets as a set of bytes objects
         # from a list of unicode objects
         targets = md.get('targets', [])
         targets = set(map(cast_bytes, targets))
 
         retries = md.get('retries', 0)
         self.retries[msg_id] = retries
 
         # time dependencies
         after = md.get('after', None)
         if after:
             after = Dependency(after)
             if after.all:
                 if after.success:
                     after = Dependency(after.difference(self.all_completed),
                                 success=after.success,
                                 failure=after.failure,
                                 all=after.all,
                     )
                 if after.failure:
                     after = Dependency(after.difference(self.all_failed),
                                 success=after.success,
                                 failure=after.failure,
                                 all=after.all,
                     )
             if after.check(self.all_completed, self.all_failed):
                 # recast as empty set, if `after` already met,
                 # to prevent unnecessary set comparisons
                 after = MET
         else:
             after = MET
 
         # location dependencies
         follow = Dependency(md.get('follow', []))
 
         timeout = md.get('timeout', None)
         if timeout:
             timeout = float(timeout)
 
         job = Job(msg_id=msg_id, raw_msg=raw_msg, idents=idents, msg=msg,
                  header=header, targets=targets, after=after, follow=follow,
                  timeout=timeout, metadata=md,
         )
         # validate and reduce dependencies:
         for dep in after,follow:
             if not dep: # empty dependency
                 continue
             # check valid:
             if msg_id in dep or dep.difference(self.all_ids):
                 self.queue_map[msg_id] = job
                 return self.fail_unreachable(msg_id, error.InvalidDependency)
             # check if unreachable:
             if dep.unreachable(self.all_completed, self.all_failed):
                 self.queue_map[msg_id] = job
                 return self.fail_unreachable(msg_id)
 
         if after.check(self.all_completed, self.all_failed):
             # time deps already met, try to run
             if not self.maybe_run(job):
                 # can't run yet
                 if msg_id not in self.all_failed:
                     # could have failed as unreachable
                     self.save_unmet(job)
         else:
             self.save_unmet(job)
 
     def job_timeout(self, job, timeout_id):
         """callback for a job's timeout.
         
         The job may or may not have been run at this point.
         """
         if job.timeout_id != timeout_id:
             # not the most recent call
             return
         now = time.time()
         if job.timeout >= (now + 1):
             self.log.warn("task %s timeout fired prematurely: %s > %s",
                 job.msg_id, job.timeout, now
             )
         if job.msg_id in self.queue_map:
             # still waiting, but ran out of time
             self.log.info("task %r timed out", job.msg_id)
             self.fail_unreachable(job.msg_id, error.TaskTimeout)
 
     def fail_unreachable(self, msg_id, why=error.ImpossibleDependency):
         """a task has become unreachable, send a reply with an ImpossibleDependency
         error."""
         if msg_id not in self.queue_map:
             self.log.error("task %r already failed!", msg_id)
             return
         job = self.queue_map.pop(msg_id)
         # lazy-delete from the queue
         job.removed = True
         for mid in job.dependents:
             if mid in self.graph:
                 self.graph[mid].remove(msg_id)
 
         try:
             raise why()
         except:
             content = error.wrap_exception()
         self.log.debug("task %r failing as unreachable with: %s", msg_id, content['ename'])
 
         self.all_done.add(msg_id)
         self.all_failed.add(msg_id)
 
         msg = self.session.send(self.client_stream, 'apply_reply', content,
                                                 parent=job.header, ident=job.idents)
         self.session.send(self.mon_stream, msg, ident=[b'outtask']+job.idents)
 
         self.update_graph(msg_id, success=False)
 
     def available_engines(self):
         """return a list of available engine indices based on HWM"""
         if not self.hwm:
             return list(range(len(self.targets)))
         available = []
         for idx in range(len(self.targets)):
             if self.loads[idx] < self.hwm:
                 available.append(idx)
         return available
 
     def maybe_run(self, job):
         """check location dependencies, and run if they are met."""
         msg_id = job.msg_id
         self.log.debug("Attempting to assign task %s", msg_id)
         available = self.available_engines()
         if not available:
             # no engines, definitely can't run
             return False
         
         if job.follow or job.targets or job.blacklist or self.hwm:
             # we need a can_run filter
             def can_run(idx):
                 # check hwm
                 if self.hwm and self.loads[idx] == self.hwm:
                     return False
                 target = self.targets[idx]
                 # check blacklist
                 if target in job.blacklist:
                     return False
                 # check targets
                 if job.targets and target not in job.targets:
                     return False
                 # check follow
                 return job.follow.check(self.completed[target], self.failed[target])
 
             indices = list(filter(can_run, available))
 
             if not indices:
                 # couldn't run
                 if job.follow.all:
                     # check follow for impossibility
                     dests = set()
                     relevant = set()
                     if job.follow.success:
                         relevant = self.all_completed
                     if job.follow.failure:
                         relevant = relevant.union(self.all_failed)
                     for m in job.follow.intersection(relevant):
                         dests.add(self.destinations[m])
                     if len(dests) > 1:
                         self.queue_map[msg_id] = job
                         self.fail_unreachable(msg_id)
                         return False
                 if job.targets:
                     # check blacklist+targets for impossibility
                     job.targets.difference_update(job.blacklist)
                     if not job.targets or not job.targets.intersection(self.targets):
                         self.queue_map[msg_id] = job
                         self.fail_unreachable(msg_id)
                         return False
                 return False
         else:
             indices = None
 
         self.submit_task(job, indices)
         return True
 
     def save_unmet(self, job):
         """Save a message for later submission when its dependencies are met."""
         msg_id = job.msg_id
         self.log.debug("Adding task %s to the queue", msg_id)
         self.queue_map[msg_id] = job
         self.queue.append(job)
         # track the ids in follow or after, but not those already finished
         for dep_id in job.after.union(job.follow).difference(self.all_done):
             if dep_id not in self.graph:
                 self.graph[dep_id] = set()
             self.graph[dep_id].add(msg_id)
         
         # schedule timeout callback
         if job.timeout:
             timeout_id = job.timeout_id = job.timeout_id + 1
             self.loop.add_timeout(time.time() + job.timeout,
                 lambda : self.job_timeout(job, timeout_id)
             )
         
 
     def submit_task(self, job, indices=None):
         """Submit a task to any of a subset of our targets."""
         if indices:
             loads = [self.loads[i] for i in indices]
         else:
             loads = self.loads
         idx = self.scheme(loads)
         if indices:
             idx = indices[idx]
         target = self.targets[idx]
         # print (target, map(str, msg[:3]))
         # send job to the engine
         self.engine_stream.send(target, flags=zmq.SNDMORE, copy=False)
         self.engine_stream.send_multipart(job.raw_msg, copy=False)
         # update load
         self.add_job(idx)
         self.pending[target][job.msg_id] = job
         # notify Hub
         content = dict(msg_id=job.msg_id, engine_id=target.decode('ascii'))
         self.session.send(self.mon_stream, 'task_destination', content=content,
                         ident=[b'tracktask',self.ident])
 
 
     #-----------------------------------------------------------------------
     # Result Handling
     #-----------------------------------------------------------------------
     
     
     @util.log_errors
     def dispatch_result(self, raw_msg):
         """dispatch method for result replies"""
         try:
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.deserialize(msg, content=False, copy=False)
             engine = idents[0]
             try:
                 idx = self.targets.index(engine)
             except ValueError:
                 pass # skip load-update for dead engines
             else:
                 self.finish_job(idx)
         except Exception:
             self.log.error("task::Invalid result: %r", raw_msg, exc_info=True)
             return
 
         md = msg['metadata']
         parent = msg['parent_header']
         if md.get('dependencies_met', True):
             success = (md['status'] == 'ok')
             msg_id = parent['msg_id']
             retries = self.retries[msg_id]
             if not success and retries > 0:
                 # failed
                 self.retries[msg_id] = retries - 1
                 self.handle_unmet_dependency(idents, parent)
             else:
                 del self.retries[msg_id]
                 # relay to client and update graph
                 self.handle_result(idents, parent, raw_msg, success)
                 # send to Hub monitor
                 self.mon_stream.send_multipart([b'outtask']+raw_msg, copy=False)
         else:
             self.handle_unmet_dependency(idents, parent)
 
     def handle_result(self, idents, parent, raw_msg, success=True):
         """handle a real task result, either success or failure"""
         # first, relay result to client
         engine = idents[0]
         client = idents[1]
         # swap_ids for ROUTER-ROUTER mirror
         raw_msg[:2] = [client,engine]
         # print (map(str, raw_msg[:4]))
         self.client_stream.send_multipart(raw_msg, copy=False)
         # now, update our data structures
         msg_id = parent['msg_id']
         self.pending[engine].pop(msg_id)
         if success:
             self.completed[engine].add(msg_id)
             self.all_completed.add(msg_id)
         else:
             self.failed[engine].add(msg_id)
             self.all_failed.add(msg_id)
         self.all_done.add(msg_id)
         self.destinations[msg_id] = engine
 
         self.update_graph(msg_id, success)
 
     def handle_unmet_dependency(self, idents, parent):
         """handle an unmet dependency"""
         engine = idents[0]
         msg_id = parent['msg_id']
 
         job = self.pending[engine].pop(msg_id)
         job.blacklist.add(engine)
 
         if job.blacklist == job.targets:
             self.queue_map[msg_id] = job
             self.fail_unreachable(msg_id)
         elif not self.maybe_run(job):
             # resubmit failed
             if msg_id not in self.all_failed:
                 # put it back in our dependency tree
                 self.save_unmet(job)
 
         if self.hwm:
             try:
                 idx = self.targets.index(engine)
             except ValueError:
                 pass # skip load-update for dead engines
             else:
                 if self.loads[idx] == self.hwm-1:
                     self.update_graph(None)
 
     def update_graph(self, dep_id=None, success=True):
         """dep_id just finished. Update our dependency
         graph and submit any jobs that just became runnable.
 
         Called with dep_id=None to update entire graph for hwm, but without finishing a task.
         """
         # print ("\n\n***********")
         # pprint (dep_id)
         # pprint (self.graph)
         # pprint (self.queue_map)
         # pprint (self.all_completed)
         # pprint (self.all_failed)
         # print ("\n\n***********\n\n")
         # update any jobs that depended on the dependency
         msg_ids = self.graph.pop(dep_id, [])
 
         # recheck *all* jobs if
         # a) we have HWM and an engine just become no longer full
         # or b) dep_id was given as None
         
         if dep_id is None or self.hwm and any( [ load==self.hwm-1 for load in self.loads ]):
             jobs = self.queue
             using_queue = True
         else:
             using_queue = False
             jobs = deque(sorted( self.queue_map[msg_id] for msg_id in msg_ids ))
         
         to_restore = []
         while jobs:
             job = jobs.popleft()
             if job.removed:
                 continue
             msg_id = job.msg_id
             
             put_it_back = True
             
             if job.after.unreachable(self.all_completed, self.all_failed)\
                     or job.follow.unreachable(self.all_completed, self.all_failed):
                 self.fail_unreachable(msg_id)
                 put_it_back = False
 
             elif job.after.check(self.all_completed, self.all_failed): # time deps met, maybe run
                 if self.maybe_run(job):
                     put_it_back = False
                     self.queue_map.pop(msg_id)
                     for mid in job.dependents:
                         if mid in self.graph:
                             self.graph[mid].remove(msg_id)
                     
                     # abort the loop if we just filled up all of our engines.
                     # avoids an O(N) operation in situation of full queue,
                     # where graph update is triggered as soon as an engine becomes
                     # non-full, and all tasks after the first are checked,
                     # even though they can't run.
                     if not self.available_engines():
                         break
             
             if using_queue and put_it_back:
                 # popped a job from the queue but it neither ran nor failed,
                 # so we need to put it back when we are done
                 # make sure to_restore preserves the same ordering
                 to_restore.append(job)
         
         # put back any tasks we popped but didn't run
         if using_queue:
             self.queue.extendleft(to_restore)
     
     #----------------------------------------------------------------------
     # methods to be overridden by subclasses
     #----------------------------------------------------------------------
 
     def add_job(self, idx):
         """Called after self.targets[idx] just got the job with header.
         Override with subclasses.  The default ordering is simple LRU.
         The default loads are the number of outstanding jobs."""
         self.loads[idx] += 1
         for lis in (self.targets, self.loads):
             lis.append(lis.pop(idx))
 
 
     def finish_job(self, idx):
         """Called after self.targets[idx] just finished a job.
         Override with subclasses."""
         self.loads[idx] -= 1
 
 
 
 def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, reg_addr, config=None,
                         logname='root', log_url=None, loglevel=logging.DEBUG,
                         identity=b'task', in_thread=False):
 
     ZMQStream = zmqstream.ZMQStream
 
     if config:
         # unwrap dict back into Config
         config = Config(config)
 
     if in_thread:
         # use instance() to get the same Context/Loop as our parent
         ctx = zmq.Context.instance()
         loop = ioloop.IOLoop.instance()
     else:
         # in a process, don't use instance()
         # for safety with multiprocessing
         ctx = zmq.Context()
         loop = ioloop.IOLoop()
     ins = ZMQStream(ctx.socket(zmq.ROUTER),loop)
     util.set_hwm(ins, 0)
     ins.setsockopt(zmq.IDENTITY, identity + b'_in')
     ins.bind(in_addr)
 
     outs = ZMQStream(ctx.socket(zmq.ROUTER),loop)
     util.set_hwm(outs, 0)
     outs.setsockopt(zmq.IDENTITY, identity + b'_out')
     outs.bind(out_addr)
     mons = zmqstream.ZMQStream(ctx.socket(zmq.PUB),loop)
     util.set_hwm(mons, 0)
     mons.connect(mon_addr)
     nots = zmqstream.ZMQStream(ctx.socket(zmq.SUB),loop)
     nots.setsockopt(zmq.SUBSCRIBE, b'')
     nots.connect(not_addr)
     
     querys = ZMQStream(ctx.socket(zmq.DEALER),loop)
     querys.connect(reg_addr)
     
     # setup logging.
     if in_thread:
         log = Application.instance().log
     else:
         if log_url:
             log = connect_logger(logname, ctx, log_url, root="scheduler", loglevel=loglevel)
         else:
             log = local_logger(logname, loglevel)
 
     scheduler = TaskScheduler(client_stream=ins, engine_stream=outs,
                             mon_stream=mons, notifier_stream=nots,
                             query_stream=querys,
                             loop=loop, log=log,
                             config=config)
     scheduler.start()
     if not in_thread:
         try:
             loop.start()
         except KeyboardInterrupt:
             scheduler.log.critical("Interrupted, exiting...")
 

IPython/parallel/tests/clienttest.py

@@ -1,192 +1,192 b''
 """base class for parallel client tests
 
 Authors:
 
 * Min RK
 """
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 from __future__ import print_function
 
 import sys
 import tempfile
 import time
 
 from nose import SkipTest
 
 import zmq
 from zmq.tests import BaseZMQTestCase
 
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 from IPython.parallel import error
 from IPython.parallel import Client
 
 from IPython.parallel.tests import launchers, add_engines
 
 # simple tasks for use in apply tests
 
 def segfault():
     """this will segfault"""
     import ctypes
     ctypes.memset(-1,0,1)
 
 def crash():
     """from stdlib crashers in the test suite"""
     import types
     if sys.platform.startswith('win'):
         import ctypes
         ctypes.windll.kernel32.SetErrorMode(0x0002);
     args = [ 0, 0, 0, 0, b'\x04\x71\x00\x00', (), (), (), '', '', 1, b'']
     if sys.version_info[0] >= 3:
         # Python3 adds 'kwonlyargcount' as the second argument to Code
         args.insert(1, 0)
         
     co = types.CodeType(*args)
     exec(co)
 
 def wait(n):
     """sleep for a time"""
     import time
     time.sleep(n)
     return n
 
 def raiser(eclass):
     """raise an exception"""
     raise eclass()
 
 def generate_output():
     """function for testing output
     
     publishes two outputs of each type, and returns
     a rich displayable object.
     """
     
     import sys
     from IPython.core.display import display, HTML, Math
     
     print("stdout")
     print("stderr", file=sys.stderr)
     
     display(HTML("<b>HTML</b>"))
     
     print("stdout2")
     print("stderr2", file=sys.stderr)
     
     display(Math(r"\alpha=\beta"))
     
     return Math("42")
 
 # test decorator for skipping tests when libraries are unavailable
 def skip_without(*names):
     """skip a test if some names are not importable"""
     @decorator
     def skip_without_names(f, *args, **kwargs):
         """decorator to skip tests in the absence of numpy."""
         for name in names:
             try:
                 __import__(name)
             except ImportError:
                 raise SkipTest
         return f(*args, **kwargs)
     return skip_without_names
 
 #-------------------------------------------------------------------------------
 # Classes
 #-------------------------------------------------------------------------------
 
 
 class ClusterTestCase(BaseZMQTestCase):
     timeout = 10
     
     def add_engines(self, n=1, block=True):
         """add multiple engines to our cluster"""
         self.engines.extend(add_engines(n))
         if block:
             self.wait_on_engines()
 
     def minimum_engines(self, n=1, block=True):
         """add engines until there are at least n connected"""
         self.engines.extend(add_engines(n, total=True))
         if block:
             self.wait_on_engines()
             
     
     def wait_on_engines(self, timeout=5):
         """wait for our engines to connect."""
         n = len(self.engines)+self.base_engine_count
         tic = time.time()
         while time.time()-tic < timeout and len(self.client.ids) < n:
             time.sleep(0.1)
         
         assert not len(self.client.ids) < n, "waiting for engines timed out"
     
     def client_wait(self, client, jobs=None, timeout=-1):
         """my wait wrapper, sets a default finite timeout to avoid hangs"""
         if timeout < 0:
             timeout = self.timeout
         return Client.wait(client, jobs, timeout)
     
     def connect_client(self):
         """connect a client with my Context, and track its sockets for cleanup"""
         c = Client(profile='iptest', context=self.context)
         c.wait = lambda *a, **kw: self.client_wait(c, *a, **kw)
         
         for name in filter(lambda n:n.endswith('socket'), dir(c)):
             s = getattr(c, name)
             s.setsockopt(zmq.LINGER, 0)
             self.sockets.append(s)
         return c
     
     def assertRaisesRemote(self, etype, f, *args, **kwargs):
         try:
             try:
                 f(*args, **kwargs)
             except error.CompositeError as e:
                 e.raise_exception()
         except error.RemoteError as e:
             self.assertEqual(etype.__name__, e.ename, "Should have raised %r, but raised %r"%(etype.__name__, e.ename))
         else:
             self.fail("should have raised a RemoteError")
             
     def _wait_for(self, f, timeout=10):
         """wait for a condition"""
         tic = time.time()
         while time.time() <= tic + timeout:
             if f():
                 return
             time.sleep(0.1)
             self.client.spin()
         if not f():
             print("Warning: Awaited condition never arrived")
     
     def setUp(self):
         BaseZMQTestCase.setUp(self)
         self.client = self.connect_client()
         # start every test with clean engine namespaces:
         self.client.clear(block=True)
         self.base_engine_count=len(self.client.ids)
         self.engines=[]
     
     def tearDown(self):
         # self.client.clear(block=True)
         # close fds:
         for e in filter(lambda e: e.poll() is not None, launchers):
             launchers.remove(e)
         
         # allow flushing of incoming messages to prevent crash on socket close
         self.client.wait(timeout=2)
         # time.sleep(2)
         self.client.spin()
         self.client.close()
         BaseZMQTestCase.tearDown(self)
         # this will be redundant when pyzmq merges PR #88
         # self.context.term()
         # print tempfile.TemporaryFile().fileno(),
         # sys.stdout.flush()
         

IPython/parallel/util.py

@@ -1,389 +1,389 b''
 """Some generic utilities for dealing with classes, urls, and serialization."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import logging
 import os
 import re
 import stat
 import socket
 import sys
 import warnings
 from signal import signal, SIGINT, SIGABRT, SIGTERM
 try:
     from signal import SIGKILL
 except ImportError:
     SIGKILL=None
 from types import FunctionType
 
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 import zmq
 from zmq.log import handlers
 
 from IPython.utils.log import get_logger
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 from IPython.config.application import Application
 from IPython.utils.localinterfaces import localhost, is_public_ip, public_ips
 from IPython.utils.py3compat import string_types, iteritems, itervalues
 from IPython.kernel.zmq.log import EnginePUBHandler
 
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class Namespace(dict):
     """Subclass of dict for attribute access to keys."""
     
     def __getattr__(self, key):
         """getattr aliased to getitem"""
         if key in self:
             return self[key]
         else:
             raise NameError(key)
 
     def __setattr__(self, key, value):
         """setattr aliased to setitem, with strict"""
         if hasattr(dict, key):
             raise KeyError("Cannot override dict keys %r"%key)
         self[key] = value
     
 
 class ReverseDict(dict):
     """simple double-keyed subset of dict methods."""
     
     def __init__(self, *args, **kwargs):
         dict.__init__(self, *args, **kwargs)
         self._reverse = dict()
         for key, value in iteritems(self):
             self._reverse[value] = key
     
     def __getitem__(self, key):
         try:
             return dict.__getitem__(self, key)
         except KeyError:
             return self._reverse[key]
     
     def __setitem__(self, key, value):
         if key in self._reverse:
             raise KeyError("Can't have key %r on both sides!"%key)
         dict.__setitem__(self, key, value)
         self._reverse[value] = key
     
     def pop(self, key):
         value = dict.pop(self, key)
         self._reverse.pop(value)
         return value
     
     def get(self, key, default=None):
         try:
             return self[key]
         except KeyError:
             return default
 
 #-----------------------------------------------------------------------------
 # Functions
 #-----------------------------------------------------------------------------
 
 @decorator
 def log_errors(f, self, *args, **kwargs):
     """decorator to log unhandled exceptions raised in a method.
     
     For use wrapping on_recv callbacks, so that exceptions
     do not cause the stream to be closed.
     """
     try:
         return f(self, *args, **kwargs)
     except Exception:
         self.log.error("Uncaught exception in %r" % f, exc_info=True)
     
 
 def is_url(url):
     """boolean check for whether a string is a zmq url"""
     if '://' not in url:
         return False
     proto, addr = url.split('://', 1)
     if proto.lower() not in ['tcp','pgm','epgm','ipc','inproc']:
         return False
     return True
 
 def validate_url(url):
     """validate a url for zeromq"""
     if not isinstance(url, string_types):
         raise TypeError("url must be a string, not %r"%type(url))
     url = url.lower()
     
     proto_addr = url.split('://')
     assert len(proto_addr) == 2, 'Invalid url: %r'%url
     proto, addr = proto_addr
     assert proto in ['tcp','pgm','epgm','ipc','inproc'], "Invalid protocol: %r"%proto
     
     # domain pattern adapted from http://www.regexlib.com/REDetails.aspx?regexp_id=391
     # author: Remi Sabourin
     pat = re.compile(r'^([\w\d]([\w\d\-]{0,61}[\w\d])?\.)*[\w\d]([\w\d\-]{0,61}[\w\d])?$')
     
     if proto == 'tcp':
         lis = addr.split(':')
         assert len(lis) == 2, 'Invalid url: %r'%url
         addr,s_port = lis
         try:
             port = int(s_port)
         except ValueError:
             raise AssertionError("Invalid port %r in url: %r"%(port, url))
         
         assert addr == '*' or pat.match(addr) is not None, 'Invalid url: %r'%url
         
     else:
         # only validate tcp urls currently
         pass
     
     return True
 
 
 def validate_url_container(container):
     """validate a potentially nested collection of urls."""
     if isinstance(container, string_types):
         url = container
         return validate_url(url)
     elif isinstance(container, dict):
         container = itervalues(container)
     
     for element in container:
         validate_url_container(element)
 
 
 def split_url(url):
     """split a zmq url (tcp://ip:port) into ('tcp','ip','port')."""
     proto_addr = url.split('://')
     assert len(proto_addr) == 2, 'Invalid url: %r'%url
     proto, addr = proto_addr
     lis = addr.split(':')
     assert len(lis) == 2, 'Invalid url: %r'%url
     addr,s_port = lis
     return proto,addr,s_port
 
 
 def disambiguate_ip_address(ip, location=None):
     """turn multi-ip interfaces '0.0.0.0' and '*' into a connectable address
     
     Explicit IP addresses are returned unmodified.
     
     Parameters
     ----------
     
     ip : IP address
         An IP address, or the special values 0.0.0.0, or *
     location: IP address, optional
         A public IP of the target machine.
         If location is an IP of the current machine,
         localhost will be returned,
         otherwise location will be returned.
     """
     if ip in {'0.0.0.0', '*'}:
         if not location:
             # unspecified location, localhost is the only choice
             ip = localhost()
         elif is_public_ip(location):
             # location is a public IP on this machine, use localhost
             ip = localhost()
         elif not public_ips():
             # this machine's public IPs cannot be determined,
             # assume `location` is not this machine
             warnings.warn("IPython could not determine public IPs", RuntimeWarning)
             ip = location
         else:
             # location is not this machine, do not use loopback
             ip = location
     return ip
 
 
 def disambiguate_url(url, location=None):
     """turn multi-ip interfaces '0.0.0.0' and '*' into connectable
     ones, based on the location (default interpretation is localhost).
     
     This is for zeromq urls, such as ``tcp://*:10101``.
     """
     try:
         proto,ip,port = split_url(url)
     except AssertionError:
         # probably not tcp url; could be ipc, etc.
         return url
     
     ip = disambiguate_ip_address(ip,location)
     
     return "%s://%s:%s"%(proto,ip,port)
 
 
 #--------------------------------------------------------------------------
 # helpers for implementing old MEC API via view.apply
 #--------------------------------------------------------------------------
 
 def interactive(f):
     """decorator for making functions appear as interactively defined.
     This results in the function being linked to the user_ns as globals()
     instead of the module globals().
     """
     
     # build new FunctionType, so it can have the right globals
     # interactive functions never have closures, that's kind of the point
     if isinstance(f, FunctionType):
         mainmod = __import__('__main__')
         f = FunctionType(f.__code__, mainmod.__dict__,
             f.__name__, f.__defaults__,
         )
     # associate with __main__ for uncanning
     f.__module__ = '__main__'
     return f
 
 @interactive
 def _push(**ns):
     """helper method for implementing `client.push` via `client.apply`"""
     user_ns = globals()
     tmp = '_IP_PUSH_TMP_'
     while tmp in user_ns:
         tmp = tmp + '_'
     try:
         for name, value in ns.items():
             user_ns[tmp] = value
             exec("%s = %s" % (name, tmp), user_ns)
     finally:
         user_ns.pop(tmp, None)
 
 @interactive
 def _pull(keys):
     """helper method for implementing `client.pull` via `client.apply`"""
     if isinstance(keys, (list,tuple, set)):
         return [eval(key, globals()) for key in keys]
     else:
         return eval(keys, globals())
 
 @interactive
 def _execute(code):
     """helper method for implementing `client.execute` via `client.apply`"""
     exec(code, globals())
 
 #--------------------------------------------------------------------------
 # extra process management utilities
 #--------------------------------------------------------------------------
 
 _random_ports = set()
 
 def select_random_ports(n):
     """Selects and return n random ports that are available."""
     ports = []
     for i in range(n):
         sock = socket.socket()
         sock.bind(('', 0))
         while sock.getsockname()[1] in _random_ports:
             sock.close()
             sock = socket.socket()
             sock.bind(('', 0))
         ports.append(sock)
     for i, sock in enumerate(ports):
         port = sock.getsockname()[1]
         sock.close()
         ports[i] = port
         _random_ports.add(port)
     return ports
 
 def signal_children(children):
     """Relay interupt/term signals to children, for more solid process cleanup."""
     def terminate_children(sig, frame):
         log = get_logger()
         log.critical("Got signal %i, terminating children..."%sig)
         for child in children:
             child.terminate()
         
         sys.exit(sig != SIGINT)
         # sys.exit(sig)
     for sig in (SIGINT, SIGABRT, SIGTERM):
         signal(sig, terminate_children)
 
 def generate_exec_key(keyfile):
     import uuid
     newkey = str(uuid.uuid4())
     with open(keyfile, 'w') as f:
         # f.write('ipython-key ')
         f.write(newkey+'\n')
     # set user-only RW permissions (0600)
     # this will have no effect on Windows
     os.chmod(keyfile, stat.S_IRUSR|stat.S_IWUSR)
 
 
 def integer_loglevel(loglevel):
     try:
         loglevel = int(loglevel)
     except ValueError:
         if isinstance(loglevel, str):
             loglevel = getattr(logging, loglevel)
     return loglevel
 
 def connect_logger(logname, context, iface, root="ip", loglevel=logging.DEBUG):
     logger = logging.getLogger(logname)
     if any([isinstance(h, handlers.PUBHandler) for h in logger.handlers]):
         # don't add a second PUBHandler
         return
     loglevel = integer_loglevel(loglevel)
     lsock = context.socket(zmq.PUB)
     lsock.connect(iface)
     handler = handlers.PUBHandler(lsock)
     handler.setLevel(loglevel)
     handler.root_topic = root
     logger.addHandler(handler)
     logger.setLevel(loglevel)
 
 def connect_engine_logger(context, iface, engine, loglevel=logging.DEBUG):
     logger = logging.getLogger()
     if any([isinstance(h, handlers.PUBHandler) for h in logger.handlers]):
         # don't add a second PUBHandler
         return
     loglevel = integer_loglevel(loglevel)
     lsock = context.socket(zmq.PUB)
     lsock.connect(iface)
     handler = EnginePUBHandler(engine, lsock)
     handler.setLevel(loglevel)
     logger.addHandler(handler)
     logger.setLevel(loglevel)
     return logger
 
 def local_logger(logname, loglevel=logging.DEBUG):
     loglevel = integer_loglevel(loglevel)
     logger = logging.getLogger(logname)
     if any([isinstance(h, logging.StreamHandler) for h in logger.handlers]):
         # don't add a second StreamHandler
         return
     handler = logging.StreamHandler()
     handler.setLevel(loglevel)
     formatter = logging.Formatter("%(asctime)s.%(msecs).03d [%(name)s] %(message)s",
                 datefmt="%Y-%m-%d %H:%M:%S")
     handler.setFormatter(formatter)
 
     logger.addHandler(handler)
     logger.setLevel(loglevel)
     return logger
 
 def set_hwm(sock, hwm=0):
     """set zmq High Water Mark on a socket
     
     in a way that always works for various pyzmq / libzmq versions.
     """
     import zmq
     
     for key in ('HWM', 'SNDHWM', 'RCVHWM'):
         opt = getattr(zmq, key, None)
         if opt is None:
             continue
         try:
             sock.setsockopt(opt, hwm)
         except zmq.ZMQError:
             pass
 
         

IPython/testing/decorators.py

@@ -1,400 +1,400 b''
 # -*- coding: utf-8 -*-
 """Decorators for labeling test objects.
 
 Decorators that merely return a modified version of the original function
 object are straightforward.  Decorators that return a new function object need
 to use nose.tools.make_decorator(original_function)(decorator) in returning the
 decorator, in order to preserve metadata such as function name, setup and
 teardown functions and so on - see nose.tools for more information.
 
 This module provides a set of useful decorators meant to be ready to use in
 your own tests.  See the bottom of the file for the ready-made ones, and if you
 find yourself writing a new one that may be of generic use, add it here.
 
 Included decorators:
 
 
 Lightweight testing that remains unittest-compatible.
 
 - An @as_unittest decorator can be used to tag any normal parameter-less
   function as a unittest TestCase.  Then, both nose and normal unittest will
   recognize it as such.  This will make it easier to migrate away from Nose if
   we ever need/want to while maintaining very lightweight tests.
 
 NOTE: This file contains IPython-specific decorators. Using the machinery in
 IPython.external.decorators, we import either numpy.testing.decorators if numpy is
 available, OR use equivalent code in IPython.external._decorators, which
 we've copied verbatim from numpy.
 
 Authors
 -------
 
 - Fernando Perez <[e-mail]>
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2009-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Stdlib imports
 import sys
 import os
 import tempfile
 import unittest
 
 # Third-party imports
 
 # This is Michele Simionato's decorator module, kept verbatim.
-from IPython.external.decorator import decorator
+from decorator import decorator
 
 # Expose the unittest-driven decorators
 from .ipunittest import ipdoctest, ipdocstring
 
 # Grab the numpy-specific decorators which we keep in a file that we
 # occasionally update from upstream: decorators.py is a copy of
 # numpy.testing.decorators, we expose all of it here.
 from IPython.external.decorators import *
 
 # For onlyif_cmd_exists decorator
 from IPython.utils.process import is_cmd_found
 from IPython.utils.py3compat import string_types
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 # Simple example of the basic idea
 def as_unittest(func):
     """Decorator to make a simple function into a normal test via unittest."""
     class Tester(unittest.TestCase):
         def test(self):
             func()
 
     Tester.__name__ = func.__name__
 
     return Tester
 
 # Utility functions
 
 def apply_wrapper(wrapper,func):
     """Apply a wrapper to a function for decoration.
 
     This mixes Michele Simionato's decorator tool with nose's make_decorator,
     to apply a wrapper in a decorator so that all nose attributes, as well as
     function signature and other properties, survive the decoration cleanly.
     This will ensure that wrapped functions can still be well introspected via
     IPython, for example.
     """
     import nose.tools
 
     return decorator(wrapper,nose.tools.make_decorator(func)(wrapper))
 
 
 def make_label_dec(label,ds=None):
     """Factory function to create a decorator that applies one or more labels.
 
     Parameters
     ----------
       label : string or sequence
       One or more labels that will be applied by the decorator to the functions
     it decorates.  Labels are attributes of the decorated function with their
     value set to True.
 
       ds : string
       An optional docstring for the resulting decorator.  If not given, a
       default docstring is auto-generated.
 
     Returns
     -------
       A decorator.
 
     Examples
     --------
 
     A simple labeling decorator:
 
     >>> slow = make_label_dec('slow')
     >>> slow.__doc__
     "Labels a test as 'slow'."
     
     And one that uses multiple labels and a custom docstring:
     
     >>> rare = make_label_dec(['slow','hard'],
     ... "Mix labels 'slow' and 'hard' for rare tests.")
     >>> rare.__doc__
     "Mix labels 'slow' and 'hard' for rare tests."
 
     Now, let's test using this one:
     >>> @rare
     ... def f(): pass
     ...
     >>>
     >>> f.slow
     True
     >>> f.hard
     True
     """
 
     if isinstance(label, string_types):
         labels = [label]
     else:
         labels = label
 
     # Validate that the given label(s) are OK for use in setattr() by doing a
     # dry run on a dummy function.
     tmp = lambda : None
     for label in labels:
         setattr(tmp,label,True)
 
     # This is the actual decorator we'll return
     def decor(f):
         for label in labels:
             setattr(f,label,True)
         return f
 
     # Apply the user's docstring, or autogenerate a basic one
     if ds is None:
         ds = "Labels a test as %r." % label
     decor.__doc__ = ds
 
     return decor
 
 
 # Inspired by numpy's skipif, but uses the full apply_wrapper utility to
 # preserve function metadata better and allows the skip condition to be a
 # callable.
 def skipif(skip_condition, msg=None):
     ''' Make function raise SkipTest exception if skip_condition is true
 
     Parameters
     ----------
 
     skip_condition : bool or callable
       Flag to determine whether to skip test. If the condition is a
       callable, it is used at runtime to dynamically make the decision. This
       is useful for tests that may require costly imports, to delay the cost
       until the test suite is actually executed.
     msg : string
       Message to give on raising a SkipTest exception.
 
     Returns
     -------
     decorator : function
       Decorator, which, when applied to a function, causes SkipTest
       to be raised when the skip_condition was True, and the function
       to be called normally otherwise.
 
     Notes
     -----
     You will see from the code that we had to further decorate the
     decorator with the nose.tools.make_decorator function in order to
     transmit function name, and various other metadata.
     '''
 
     def skip_decorator(f):
         # Local import to avoid a hard nose dependency and only incur the
         # import time overhead at actual test-time.
         import nose
 
         # Allow for both boolean or callable skip conditions.
         if callable(skip_condition):
             skip_val = skip_condition
         else:
             skip_val = lambda : skip_condition
 
         def get_msg(func,msg=None):
             """Skip message with information about function being skipped."""
             if msg is None: out = 'Test skipped due to test condition.'
             else: out = msg
             return "Skipping test: %s. %s" % (func.__name__,out)
 
         # We need to define *two* skippers because Python doesn't allow both
         # return with value and yield inside the same function.
         def skipper_func(*args, **kwargs):
             """Skipper for normal test functions."""
             if skip_val():
                 raise nose.SkipTest(get_msg(f,msg))
             else:
                 return f(*args, **kwargs)
 
         def skipper_gen(*args, **kwargs):
             """Skipper for test generators."""
             if skip_val():
                 raise nose.SkipTest(get_msg(f,msg))
             else:
                 for x in f(*args, **kwargs):
                     yield x
 
         # Choose the right skipper to use when building the actual generator.
         if nose.util.isgenerator(f):
             skipper = skipper_gen
         else:
             skipper = skipper_func
 
         return nose.tools.make_decorator(f)(skipper)
 
     return skip_decorator
 
 # A version with the condition set to true, common case just to attach a message
 # to a skip decorator
 def skip(msg=None):
     """Decorator factory - mark a test function for skipping from test suite.
 
     Parameters
     ----------
       msg : string
         Optional message to be added.
 
     Returns
     -------
        decorator : function
          Decorator, which, when applied to a function, causes SkipTest
          to be raised, with the optional message added.
       """
 
     return skipif(True,msg)
 
 
 def onlyif(condition, msg):
     """The reverse from skipif, see skipif for details."""
 
     if callable(condition):
         skip_condition = lambda : not condition()
     else:
         skip_condition = lambda : not condition
 
     return skipif(skip_condition, msg)
 
 #-----------------------------------------------------------------------------
 # Utility functions for decorators
 def module_not_available(module):
     """Can module be imported?  Returns true if module does NOT import.
 
     This is used to make a decorator to skip tests that require module to be
     available, but delay the 'import numpy' to test execution time.
     """
     try:
         mod = __import__(module)
         mod_not_avail = False
     except ImportError:
         mod_not_avail = True
 
     return mod_not_avail
 
 
 def decorated_dummy(dec, name):
     """Return a dummy function decorated with dec, with the given name.
     
     Examples
     --------
     import IPython.testing.decorators as dec
     setup = dec.decorated_dummy(dec.skip_if_no_x11, __name__)
     """
     dummy = lambda: None
     dummy.__name__ = name
     return dec(dummy)
 
 #-----------------------------------------------------------------------------
 # Decorators for public use
 
 # Decorators to skip certain tests on specific platforms.
 skip_win32 = skipif(sys.platform == 'win32',
                     "This test does not run under Windows")
 skip_linux = skipif(sys.platform.startswith('linux'),
                     "This test does not run under Linux")
 skip_osx = skipif(sys.platform == 'darwin',"This test does not run under OS X")
 
 
 # Decorators to skip tests if not on specific platforms.
 skip_if_not_win32 = skipif(sys.platform != 'win32',
                            "This test only runs under Windows")
 skip_if_not_linux = skipif(not sys.platform.startswith('linux'),
                            "This test only runs under Linux")
 skip_if_not_osx = skipif(sys.platform != 'darwin',
                          "This test only runs under OSX")
 
 
 _x11_skip_cond = (sys.platform not in ('darwin', 'win32') and
                   os.environ.get('DISPLAY', '') == '')
 _x11_skip_msg = "Skipped under *nix when X11/XOrg not available"
 
 skip_if_no_x11 = skipif(_x11_skip_cond, _x11_skip_msg)
 
 # not a decorator itself, returns a dummy function to be used as setup
 def skip_file_no_x11(name):
     return decorated_dummy(skip_if_no_x11, name) if _x11_skip_cond else None
 
 # Other skip decorators
 
 # generic skip without module
 skip_without = lambda mod: skipif(module_not_available(mod), "This test requires %s" % mod)
 
 skipif_not_numpy = skip_without('numpy')
 
 skipif_not_matplotlib = skip_without('matplotlib')
 
 skipif_not_sympy = skip_without('sympy')
 
 skip_known_failure = knownfailureif(True,'This test is known to fail')
 
 known_failure_py3 = knownfailureif(sys.version_info[0] >= 3, 
                                     'This test is known to fail on Python 3.')
 
 # A null 'decorator', useful to make more readable code that needs to pick
 # between different decorators based on OS or other conditions
 null_deco = lambda f: f
 
 # Some tests only run where we can use unicode paths. Note that we can't just
 # check os.path.supports_unicode_filenames, which is always False on Linux.
 try:
     f = tempfile.NamedTemporaryFile(prefix=u"tmp€")
 except UnicodeEncodeError:
     unicode_paths = False
 else:
     unicode_paths = True
     f.close()
 
 onlyif_unicode_paths = onlyif(unicode_paths, ("This test is only applicable "
                                     "where we can use unicode in filenames."))
 
 
 def onlyif_cmds_exist(*commands):
     """
     Decorator to skip test when at least one of `commands` is not found.
     """
     for cmd in commands:
         try:
             if not is_cmd_found(cmd):
                 return skip("This test runs only if command '{0}' "
                             "is installed".format(cmd))
         except ImportError as e:
             # is_cmd_found uses pywin32 on windows, which might not be available
             if sys.platform == 'win32' and 'pywin32' in str(e):
                 return skip("This test runs only if pywin32 and command '{0}' "
                             "is installed".format(cmd))
             raise e
     return null_deco
 
 def onlyif_any_cmd_exists(*commands):
     """
     Decorator to skip test unless at least one of `commands` is found.
     """
     for cmd in commands:
         try:
             if is_cmd_found(cmd):
                 return null_deco
         except ImportError as e:
             # is_cmd_found uses pywin32 on windows, which might not be available
             if sys.platform == 'win32' and 'pywin32' in str(e):
                 return skip("This test runs only if pywin32 and commands '{0}' "
                             "are installed".format(commands))
             raise e
     return skip("This test runs only if one of the commands {0} "
                 "is installed".format(commands))

setup.py

@@ -1,344 +1,345 b''
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 """Setup script for IPython.
 
 Under Posix environments it works like a typical setup.py script.
 Under Windows, the command sdist is not supported, since IPython
 requires utilities which are not available under Windows."""
 
 #-----------------------------------------------------------------------------
 #  Copyright (c) 2008-2011, IPython Development Team.
 #  Copyright (c) 2001-2007, Fernando Perez <[e-mail]>
 #  Copyright (c) 2001, Janko Hauser <[e-mail]>
 #  Copyright (c) 2001, Nathaniel Gray <[e-mail]>
 #
 #  Distributed under the terms of the Modified BSD License.
 #
 #  The full license is in the file COPYING.rst, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Minimal Python version sanity check
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 import sys
 
 # This check is also made in IPython/__init__, don't forget to update both when
 # changing Python version requirements.
 v = sys.version_info
 if v[:2] < (2,7) or (v[0] >= 3 and v[:2] < (3,3)):
     error = "ERROR: IPython requires Python version 2.7 or 3.3 or above."
     print(error, file=sys.stderr)
     sys.exit(1)
 
 PY3 = (sys.version_info[0] >= 3)
 
 # At least we're on the python version we need, move on.
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 # Stdlib imports
 import os
 import shutil
 
 from glob import glob
 
 # BEFORE importing distutils, remove MANIFEST. distutils doesn't properly
 # update it when the contents of directories change.
 if os.path.exists('MANIFEST'): os.remove('MANIFEST')
 
 from distutils.core import setup
 
 # Our own imports
 from setupbase import target_update
 
 from setupbase import (
     setup_args,
     find_packages,
     find_package_data,
     check_package_data_first,
     find_entry_points,
     build_scripts_entrypt,
     find_data_files,
     check_for_dependencies,
     git_prebuild,
     check_submodule_status,
     update_submodules,
     require_submodules,
     UpdateSubmodules,
     get_bdist_wheel,
     CompileCSS,
     JavascriptVersion,
     css_js_prerelease,
     install_symlinked,
     install_lib_symlink,
     install_scripts_for_symlink,
     unsymlink,
 )
 from setupext import setupext
 
 isfile = os.path.isfile
 pjoin = os.path.join
 
 #-------------------------------------------------------------------------------
 # Handle OS specific things
 #-------------------------------------------------------------------------------
 
 if os.name in ('nt','dos'):
     os_name = 'windows'
 else:
     os_name = os.name
 
 # Under Windows, 'sdist' has not been supported.  Now that the docs build with
 # Sphinx it might work, but let's not turn it on until someone confirms that it
 # actually works.
 if os_name == 'windows' and 'sdist' in sys.argv:
     print('The sdist command is not available under Windows.  Exiting.')
     sys.exit(1)
 
 #-------------------------------------------------------------------------------
 # Make sure we aren't trying to run without submodules
 #-------------------------------------------------------------------------------
 here = os.path.abspath(os.path.dirname(__file__))
 
 def require_clean_submodules():
     """Check on git submodules before distutils can do anything
 
     Since distutils cannot be trusted to update the tree
     after everything has been set in motion,
     this is not a distutils command.
     """
     # PACKAGERS: Add a return here to skip checks for git submodules
     
     # don't do anything if nothing is actually supposed to happen
     for do_nothing in ('-h', '--help', '--help-commands', 'clean', 'submodule'):
         if do_nothing in sys.argv:
             return
 
     status = check_submodule_status(here)
 
     if status == "missing":
         print("checking out submodules for the first time")
         update_submodules(here)
     elif status == "unclean":
         print('\n'.join([
             "Cannot build / install IPython with unclean submodules",
             "Please update submodules with",
             "    python setup.py submodule",
             "or",
             "    git submodule update",
             "or commit any submodule changes you have made."
         ]))
         sys.exit(1)
 
 require_clean_submodules()
 
 #-------------------------------------------------------------------------------
 # Things related to the IPython documentation
 #-------------------------------------------------------------------------------
 
 # update the manuals when building a source dist
 if len(sys.argv) >= 2 and sys.argv[1] in ('sdist','bdist_rpm'):
 
     # List of things to be updated. Each entry is a triplet of args for
     # target_update()
     to_update = [
                   # FIXME - Disabled for now: we need to redo an automatic way
                   # of generating the magic info inside the rst.
                   #('docs/magic.tex',
                   #['IPython/Magic.py'],
                   #"cd doc && ./update_magic.sh" ),
 
                  ('docs/man/ipcluster.1.gz',
                   ['docs/man/ipcluster.1'],
                   'cd docs/man && gzip -9c ipcluster.1 > ipcluster.1.gz'),
 
                  ('docs/man/ipcontroller.1.gz',
                   ['docs/man/ipcontroller.1'],
                   'cd docs/man && gzip -9c ipcontroller.1 > ipcontroller.1.gz'),
 
                  ('docs/man/ipengine.1.gz',
                   ['docs/man/ipengine.1'],
                   'cd docs/man && gzip -9c ipengine.1 > ipengine.1.gz'),
 
                  ('docs/man/ipython.1.gz',
                   ['docs/man/ipython.1'],
                   'cd docs/man && gzip -9c ipython.1 > ipython.1.gz'),
 
                  ]
 
 
     [ target_update(*t) for t in to_update ]
 
 #---------------------------------------------------------------------------
 # Find all the packages, package data, and data_files
 #---------------------------------------------------------------------------
 
 packages = find_packages()
 package_data = find_package_data()
 
 data_files = find_data_files()
 
 setup_args['packages'] = packages
 setup_args['package_data'] = package_data
 setup_args['data_files'] = data_files
 
 #---------------------------------------------------------------------------
 # custom distutils commands
 #---------------------------------------------------------------------------
 # imports here, so they are after setuptools import if there was one
 from distutils.command.sdist import sdist
 from distutils.command.upload import upload
 
 class UploadWindowsInstallers(upload):
 
     description = "Upload Windows installers to PyPI (only used from tools/release_windows.py)"
     user_options = upload.user_options + [
         ('files=', 'f', 'exe file (or glob) to upload')
     ]
     def initialize_options(self):
         upload.initialize_options(self)
         meta = self.distribution.metadata
         base = '{name}-{version}'.format(
             name=meta.get_name(),
             version=meta.get_version()
         )
         self.files = os.path.join('dist', '%s.*.exe' % base)
 
     def run(self):
         for dist_file in glob(self.files):
             self.upload_file('bdist_wininst', 'any', dist_file)
 
 setup_args['cmdclass'] = {
     'build_py': css_js_prerelease(
             check_package_data_first(git_prebuild('IPython'))),
     'sdist' : css_js_prerelease(git_prebuild('IPython', sdist)),
     'upload_wininst' : UploadWindowsInstallers,
     'submodule' : UpdateSubmodules,
     'css' : CompileCSS,
     'symlink': install_symlinked,
     'install_lib_symlink': install_lib_symlink,
     'install_scripts_sym': install_scripts_for_symlink,
     'unsymlink': unsymlink,
     'jsversion' : JavascriptVersion,
 }
 
 #---------------------------------------------------------------------------
 # Handle scripts, dependencies, and setuptools specific things
 #---------------------------------------------------------------------------
 
 # For some commands, use setuptools.  Note that we do NOT list install here!
 # If you want a setuptools-enhanced install, just run 'setupegg.py install'
 needs_setuptools = set(('develop', 'release', 'bdist_egg', 'bdist_rpm',
            'bdist', 'bdist_dumb', 'bdist_wininst', 'bdist_wheel',
            'egg_info', 'easy_install', 'upload', 'install_egg_info',
             ))
 
 if len(needs_setuptools.intersection(sys.argv)) > 0:
     import setuptools
 
 # This dict is used for passing extra arguments that are setuptools
 # specific to setup
 setuptools_extra_args = {}
 
 # setuptools requirements
 
 pyzmq = 'pyzmq>=13'
 
 extras_require = dict(
     parallel = [pyzmq],
     qtconsole = [pyzmq, 'pygments'],
     doc = ['Sphinx>=1.1', 'numpydoc'],
     test = ['nose>=0.10.1', 'requests'],
     terminal = [],
     nbformat = ['jsonschema>=2.0'],
     notebook = ['tornado>=4.0', pyzmq, 'jinja2', 'pygments', 'mistune>=0.5'],
     nbconvert = ['pygments', 'jinja2', 'mistune>=0.3.1']
 )
 
 if not sys.platform.startswith('win'):
     extras_require['notebook'].append('terminado>=0.3.3')
 
 if sys.version_info < (3, 3):
     extras_require['test'].append('mock')
 
 extras_require['notebook'].extend(extras_require['nbformat'])
 extras_require['nbconvert'].extend(extras_require['nbformat'])
 
 install_requires = [
+    'decorator',
     'path.py', # required by pickleshare, remove when pickleshare is added here
 ]
 
 # add readline
 if sys.platform == 'darwin':
     if 'bdist_wheel' in sys.argv[1:] or not setupext.check_for_readline():
         install_requires.append('gnureadline')
 elif sys.platform.startswith('win'):
     extras_require['terminal'].append('pyreadline>=2.0')
 
 everything = set()
 for deps in extras_require.values():
     everything.update(deps)
 extras_require['all'] = everything
 
 if 'setuptools' in sys.modules:
     # setup.py develop should check for submodules
     from setuptools.command.develop import develop
     setup_args['cmdclass']['develop'] = require_submodules(develop)
     setup_args['cmdclass']['bdist_wheel'] = css_js_prerelease(get_bdist_wheel())
     
     setuptools_extra_args['zip_safe'] = False
     setuptools_extra_args['entry_points'] = {
         'console_scripts': find_entry_points(),
         'pygments.lexers': [
             'ipythonconsole = IPython.lib.lexers:IPythonConsoleLexer',
             'ipython = IPython.lib.lexers:IPythonLexer',
             'ipython3 = IPython.lib.lexers:IPython3Lexer',
         ],
     }
     setup_args['extras_require'] = extras_require
     requires = setup_args['install_requires'] = install_requires
 
     # Script to be run by the windows binary installer after the default setup
     # routine, to add shortcuts and similar windows-only things.  Windows
     # post-install scripts MUST reside in the scripts/ dir, otherwise distutils
     # doesn't find them.
     if 'bdist_wininst' in sys.argv:
         if len(sys.argv) > 2 and \
                ('sdist' in sys.argv or 'bdist_rpm' in sys.argv):
             print("ERROR: bdist_wininst must be run alone. Exiting.", file=sys.stderr)
             sys.exit(1)
         setup_args['data_files'].append(
             ['Scripts', ('scripts/ipython.ico', 'scripts/ipython_nb.ico')])
         setup_args['scripts'] = [pjoin('scripts','ipython_win_post_install.py')]
         setup_args['options'] = {"bdist_wininst":
                                  {"install_script":
                                   "ipython_win_post_install.py"}}
 
 else:
     # If we are installing without setuptools, call this function which will
     # check for dependencies an inform the user what is needed.  This is
     # just to make life easy for users.
     for install_cmd in ('install', 'symlink'):
         if install_cmd in sys.argv:
             check_for_dependencies()
             break
     # scripts has to be a non-empty list, or install_scripts isn't called
     setup_args['scripts'] = [e.split('=')[0].strip() for e in find_entry_points()]
 
     setup_args['cmdclass']['build_scripts'] = build_scripts_entrypt
 
 #---------------------------------------------------------------------------
 # Do the actual setup now
 #---------------------------------------------------------------------------
 
 setup_args.update(setuptools_extra_args)
 
 def main():
     setup(**setup_args)
 
 if __name__ == '__main__':
     main()