def _default_pin_factory(name=os.getenv('GPIOZERO_PIN_FACTORY', None)):
    group = 'gpiozero_pin_factories'
    if name is None:
        # If no factory is explicitly specified, try various names in
        # "preferred" order. Note that in this case we only select from
        # gpiozero distribution so without explicitly specifying a name (via
        # the environment) it's impossible to auto-select a factory from
        # outside the base distribution
        #
        # We prefer RPi.GPIO here as it supports PWM, and all Pi revisions. If
        # no third-party libraries are available, however, we fall back to a
        # pure Python implementation which supports platforms like PyPy
        dist = pkg_resources.get_distribution('gpiozero')
        for name in ('rpigpio', 'rpio', 'pigpio', 'native'):
            try:
                return pkg_resources.load_entry_point(dist, group, name)()
            except Exception as e:
                warnings.warn(
                    PinFactoryFallback(
                        'Falling back from %s: %s' % (name, str(e))))
        raise BadPinFactory('Unable to load any default pin factory!')
    else:
        # Try with the name verbatim first. If that fails, attempt with the
        # lower-cased name (this ensures compatibility names work but we're
        # still case insensitive for all factories)
        for factory in pkg_resources.iter_entry_points(group, name):
            return factory.load()()
        for factory in pkg_resources.iter_entry_points(group, name.lower()):
            return factory.load()()
        raise BadPinFactory('Unable to find pin factory "%s"' % name)

    def load_external_theme(self, name):
        # type: (unicode) -> None
        """Try to load a theme using entry_points.

        Sphinx refers to ``sphinx_themes`` entry_points.
        """
        # look up for new styled entry_points at first
        entry_points = pkg_resources.iter_entry_points('sphinx.html_themes', name)
        try:
            entry_point = next(entry_points)
            self.app.registry.load_extension(self.app, entry_point.module_name)
            return
        except StopIteration:
            pass

        # look up for old styled entry_points
        for entry_point in pkg_resources.iter_entry_points('sphinx_themes'):
            target = entry_point.load()
            if callable(target):
                themedir = target()
                if not isinstance(themedir, string_types):
                    logger.warning(__('Theme extension %r does not respond correctly.') %
                                   entry_point.module_name)
            else:
                themedir = target

            themes = self.find_themes(themedir)
            for entry, theme in iteritems(themes):
                if name == entry:
                    warnings.warn('``sphinx_themes`` entry point is now deprecated. '
                                  'Please use ``sphinx.html_themes`` instead.',
                                  RemovedInSphinx20Warning)
                    self.themes[name] = theme

def _fetch():
    # get blueprints, dists, and so on from pkg_resources.
    #
    # We're careful to load all of the blueprints exactly once and before
    # registering any of them, as this ensures everything is imported before
    # any of the @bp.register-decorated methods are called
    global _blueprints
    global _distributions

    if not _distributions:
        _distributions = {}
        for dist in pkg_resources.WorkingSet():
            dist.relengapi_metadata = {}
            _distributions[dist.key] = dist

    if not _blueprints:
        _blueprints = []
        entry_points = (list(pkg_resources.iter_entry_points('relengapi_blueprints')) +
                        list(pkg_resources.iter_entry_points('relengapi.blueprints')))
        for ep in entry_points:
            bp = ep.load()
            # make sure we have only one copy of each Distribution
            bp.dist = _distributions[ep.dist.key]
            _blueprints.append(bp)

    # look for relengapi metadata for every dist containing a blueprint
    blueprint_dists = {bp.dist.key: bp.dist for bp in _blueprints}.values()
    for dist in blueprint_dists:
        ep = pkg_resources.get_entry_info(dist, 'relengapi.metadata', dist.key)
        if not ep:
            continue
        dist.relengapi_metadata = ep.load()

def install_plugins(settings):
    from sentry.plugins import register
    # entry_points={
    #    'sentry.plugins': [
    #         'phabricator = sentry_phabricator.plugins:PhabricatorPlugin'
    #     ],
    # },
    installed_apps = list(settings.INSTALLED_APPS)
    for ep in pkg_resources.iter_entry_points('sentry.apps'):
        try:
            plugin = ep.load()
        except Exception:
            import sys
            import traceback

            print >> sys.stderr, "Failed to load app %r:\n%s" % (ep.name, traceback.format_exc())
        else:
            installed_apps.append(ep.module_name)
    settings.INSTALLED_APPS = tuple(installed_apps)

    for ep in pkg_resources.iter_entry_points('sentry.plugins'):
        try:
            plugin = ep.load()
        except Exception:
            import sys
            import traceback

            print >> sys.stderr, "Failed to load plugin %r:\n%s" % (ep.name, traceback.format_exc())
        else:
            register(plugin)

def load_tests(loader, tests, pattern):
    for entry in pkg_resources.iter_entry_points('qubes.tests.extra'):
        try:
            for test_case in entry.load()():
                tests.addTests(loader.loadTestsFromTestCase(test_case))
        except Exception as err:  # pylint: disable=broad-except
            def runTest(self):
                raise err
            ExtraLoadFailure = type('ExtraLoadFailure',
                (qubes.tests.QubesTestCase,),
                {entry.name: runTest})
            tests.addTest(ExtraLoadFailure(entry.name))

    for entry in pkg_resources.iter_entry_points(
            'qubes.tests.extra.for_template'):
        try:
            for test_case in entry.load()():
                for template in qubes.tests.list_templates():
                    tests.addTests(loader.loadTestsFromTestCase(
                        type(
                            '{}_{}_{}'.format(
                                entry.name, test_case.__name__, template),
                            (test_case,),
                            {'template': template}
                        )
                    ))
        except Exception as err:  # pylint: disable=broad-except
            def runTest(self):
                raise err
            ExtraForTemplateLoadFailure = type('ExtraForTemplateLoadFailure',
                (qubes.tests.QubesTestCase,),
                {entry.name: runTest})
            tests.addTest(ExtraForTemplateLoadFailure(entry.name))

    return tests

def main(args=None):
    parser = argparse.ArgumentParser(
        description="Releng API Command Line Tool")
    parser.add_argument("--quiet", '-q', action='store_true',
                        help="Silence all logging below WARNING level")
    subparsers = parser.add_subparsers(help='sub-command help')

    # load each of the blueprints; this defines the subcommand classes.  Note that
    # create_app does this again.
    for ep in (list(pkg_resources.iter_entry_points('relengapi_blueprints')) +
               list(pkg_resources.iter_entry_points('relengapi.blueprints'))):
        ep.load()

    subcommands = [cls() for cls in Subcommand.__subclasses__()]
    for subcommand in subcommands:
        subparser = subcommand.make_parser(subparsers)
        subparser.set_defaults(_subcommand=subcommand)

    args = parser.parse_args(args)

    if args._subcommand and args._subcommand.want_logging:
        setupConsoleLogging(args.quiet)

    app = relengapi.app.create_app(cmdline=True)
    with app.app_context():
        args._subcommand.run(parser, args)

def py_annotator(verbose=False):
    """
    find python keyword plugins and update to dicts
    
Accept args:
    verbose:
        show detail message, default: False

    'verbose' argument is only for debug(will generate too mush messages).
    """
    # tw plugin
    for entrypoints in pkg_resources.iter_entry_points("zhpy.twdict"):
        tool = entrypoints.load()
        if verbose:
            print tool.title
        merger(tool.keyword, use_dict=twdict, verbose=verbose)
    merger(twdict, verbose=verbose)
    
    # cn plugin
    for entrypoints in pkg_resources.iter_entry_points("zhpy.cndict"):
        tool = entrypoints.load()
        if verbose:
            print tool.title
        merger(tool.keyword, use_dict=cndict, verbose=verbose)
    merger(cndict, verbose=verbose)

def main(argv=None):
    argv = argv if argv is not None else sys.argv[1:]
    args = manual_argument_parsing(argv)

    # Register patches
    callables = get_entry_callables(
        args.all, args.patches,
        tuple(pkg_resources.iter_entry_points('pymonkey')),
        attr='pymonkey_patch',
    )
    hook = PymonkeyImportHook(callables)
    # Important to insert at the beginning to be ahead of the stdlib importer
    sys.meta_path.insert(0, hook)

    # Allow hooks to do argument parsing
    argv_callables = get_entry_callables(
        args.all, args.patches,
        tuple(pkg_resources.iter_entry_points('pymonkey.argparse')),
        attr='pymonkey_argparse',
    )
    cmd, rest = args.cmd[0], tuple(args.cmd[1:])
    for entry_name, argv_callable in argv_callables.items():
        args, rest = tuple(argv_callable(rest))
        hook.set_entry_data(entry_name, args)

    # Call the thing
    entry, = tuple(pkg_resources.iter_entry_points('console_scripts', cmd))
    sys.argv = [cmd] + list(rest)
    return entry.load()()

def workflow_entry_points():
    """Return an iterator over all example workflows.
    """
    default = default_entry_point()
    return chain([default],
                 pkg_resources.iter_entry_points("orange.widgets.tutorials"),
                 pkg_resources.iter_entry_points("orange.widgets.workflows"))

 def test_customize_via_pkgutil_entry_point(self):
     self.forge.replace(pkg_resources, "iter_entry_points")
     entry_point = self.forge.create_wildcard_mock()
     pkg_resources.iter_entry_points("shakedown.site.customize").and_return(iter([entry_point]))
     entry_point.load().and_return(self.get_customization_function())
     self.forge.replay()
     self.assert_customization_loaded()

def trigger():
    """Trigger the primary hooks: ``sitehooks`` and ``sitecustomize``.
    
    This function may be called several times, but will only actually
    trigger the hooks once.

    """

    global triggered
    if triggered:
        return
    triggered = True

    entry_points = []
    entry_points.extend(pkg_resources.iter_entry_points('sitehooks'))
    entry_points.extend(pkg_resources.iter_entry_points('sitecustomize'))
    entry_points.sort(key=lambda ep: ep.name)

    for entry_point in entry_points:
        try:
            func = entry_point.load()
            func()
        except Exception as e:
            warnings.warn('%s during sitehook %s: %s\n%s' % (
                e.__class__.__name__,
                entry_point,
                e,
                traceback.format_exc(),
            ))

    def load_paths(self):
        """ Load the names and paths of all moksha applications and widgets.

        We must do this before actually loading the widgets or applications, to
        ensure that we parse and load each of their configuration files
        beforehand.
        """
        for app_entry in pkg_resources.iter_entry_points('moksha.application'):
            if app_entry.name in moksha._apps:
                raise MokshaException('Duplicate application name: %s' %
                                      app_entry.name)
            app_path = app_entry.dist.location
            moksha._apps[app_entry.name] = {
                    'name': app_entry.name,
                    'project_name': app_entry.dist.project_name,
                    'path': app_path,
                    }
        for widget_entry in pkg_resources.iter_entry_points('moksha.widget'):
            if widget_entry.name in moksha._widgets:
                raise MokshaException('Duplicate widget name: %s' %
                                      widget_entry.name)
            widget_path = widget_entry.dist.location
            moksha._widgets[widget_entry.name] = {
                    'name': widget_entry.name,
                    'project_name': widget_entry.dist.project_name,
                    'path': widget_path,
                    }

    def initialise_options( self, options, default_options, profiles, dependencies ):
        options['default_options'] = default_options or {}
        # env.AddMethod( self.get_option, "get_option" )
        cuppa.core.base_options.add_base_options()
        cuppa.modules.registration.add_options( self.toolchains_key )
        cuppa.modules.registration.add_options( self.dependencies_key )
        cuppa.modules.registration.add_options( self.profiles_key )
        cuppa.modules.registration.add_options( self.project_generators_key )
        cuppa.modules.registration.add_options( self.methods_key )

        for method_plugin in pkg_resources.iter_entry_points( group='cuppa.method.plugins', name=None ):
            try:
                method_plugin.load().add_options( SCons.Script.AddOption )
            except AttributeError:
                pass

        if profiles:
            for profile in profiles:
                profile.add_options( SCons.Script.AddOption )

        for profile_plugin in pkg_resources.iter_entry_points( group='cuppa.profile.plugins', name=None ):
            try:
                profile_plugin.load().add_options( SCons.Script.AddOption )
            except AttributeError:
                pass

        if dependencies:
            for dependency in dependencies:
                dependency.add_options( SCons.Script.AddOption )

        for dependency_plugin in pkg_resources.iter_entry_points( group='cuppa.dependency.plugins', name=None ):
            try:
                dependency_plugin.load().add_options( SCons.Script.AddOption )
            except AttributeError:
                pass

	def to_foreign(self, obj, name, value):  # pylint:disable=unused-argument
		"""Transform to a MongoDB-safe value."""
		
		namespace = self.namespace
		
		try:
			explicit = self.explicit
		except AttributeError:
			explicit = not namespace
		
		if not isinstance(value, (str, unicode)):
			value = canon(value)
		
		if namespace and ':' in value:  # Try to reduce to a known plugin short name.
			for point in iter_entry_points(namespace):  # TODO: Isolate.
				qualname = point.module_name
				
				if point.attrs:
					qualname += ':' + '.'.join(point.attrs)
				
				if qualname == value:
					value = point.name
					break
		
		if ':' in value:
			if not explicit:
				raise ValueError("Explicit object references not allowed.")
			
			return value
		
		if namespace and value not in (i.name for i in iter_entry_points(namespace)):
			raise ValueError('Unknown plugin "' + value + '" for namespace "' + namespace + '".')
		
		return value

def load_plugins():  # pragma: no cover
    """Load plugins with groups: isbnlib.metadata & isbnlib.formatters."""
    # get metadata plugins from entry_points
    if options.get('LOAD_METADATA_PLUGINS', True):
        try:
            for entry in iter_entry_points(group='isbnlib.metadata'):
                add_service(entry.name, entry.load())
        except Exception:
            pass
    global PROVIDERS
    _buf = list(services.keys())
    _buf.remove('default')
    PROVIDERS = sorted(_buf)
    # get formatters from entry_points
    if options.get('LOAD_FORMATTER_PLUGINS', True):
        try:
            for entry in iter_entry_points(group='isbnlib.formatters'):
                add_bibformatter(entry.name, entry.load())
        except Exception:
            pass
    global BIBFORMATS
    _buf = list(bibformatters.keys())
    _buf.remove('labels')
    _buf.remove('default')
    BIBFORMATS = sorted(_buf)

    def __init__(self, am_conf):
        self.context = dict()
        self.attribute_fetcher = dict()

        for plugin_name in worker_config.get('ACTION_PLUGINS', []):
            module_name = 'eduid_am.ams.{}'.format(plugin_name)
            try:
                plugin_module = import_module(module_name)
            except ImportError as exc:
                logger.warn('Configured plugin {} missing from sys.path (could not import {}): {}'.format(
                    plugin_name, module_name, exc))
                logger.debug('Extra debug: path: {}'.format(sys.path))
                continue
            logger.info('Registering action plugin: {} (module {})'.format(plugin_name, module_name))

            plugin_init = getattr(plugin_module, 'plugin_init')
            self.context[plugin_name] = plugin_init(am_conf)

            attr_fetcher = getattr(plugin_module, 'attribute_fetcher')
            self.attribute_fetcher[plugin_name] = attr_fetcher

        for entry_point in iter_entry_points('eduid_am.plugin_init'):
            if entry_point.name in self.context:
                logger.warn('Duplicate plugin_init entry point: {!r}'.format(entry_point.name))
            else:
                logger.debug('Calling plugin_init entry point for {!r}'.format(entry_point.name))
                plugin_init = entry_point.load()
                self.context[entry_point.name] = plugin_init(am_conf)

        for entry_point in iter_entry_points('eduid_am.attribute_fetcher'):
            if entry_point.name in self.attribute_fetcher:
                logger.warn('Duplicate attribute_fetcher entry point: {!r}'.format(entry_point.name))
            else:
                logger.debug('Registering attribute_fetcher entry point for {!r}'.format(entry_point.name))
                self.attribute_fetcher[entry_point.name] = entry_point.load()

def main(args=None):
    parser = argparse.ArgumentParser(
        description="Releng API Command Line Tool")
    parser.add_argument("--quiet", '-q', action='store_true',
                        help="Silence all logging below WARNING level")
    subparsers = parser.add_subparsers(help='sub-command help')

    # load each of the blueprints; this defines the subcommand classes.  Note that
    # create_app does this again.
    for ep in (list(pkg_resources.iter_entry_points('relengapi_blueprints')) +
               list(pkg_resources.iter_entry_points('relengapi.blueprints'))):
        ep.load()

    subcommands = [cls() for cls in Subcommand.__subclasses__()]
    for subcommand in subcommands:
        subparser = subcommand.make_parser(subparsers)
        subparser.set_defaults(_subcommand=subcommand)

    args = parser.parse_args(args)

    if args._subcommand and args._subcommand.want_logging:
        setupConsoleLogging(args.quiet)

    # make the RELENGAPI_SETTINGS env var an absolute path; without this, Flask
    # uses the application's root_dir, which isn't especially helpful in a
    # development context.
    var_name = 'RELENGAPI_SETTINGS'
    if var_name in os.environ:
        os.environ[var_name] = os.path.abspath(os.environ[var_name])

    app = relengapi.app.create_app(cmdline=True)
    with app.app_context():
        args._subcommand.run(parser, args)

    def __init__(self):
        self.env = env = Environment.getInstance()
        self.log = logging.getLogger(__name__)
        self.log.debug("initializing asterisk number resolver")

        self.__default_image = Image.open(pkg_resources.resource_filename("amires", "data/phone.png"))

        # Load resolver
        for entry in pkg_resources.iter_entry_points("phone.resolver"):
            module = entry.load()
            self.log.debug("loading resolver module '%s'" % module.__name__)
            obj = module()
            self.resolver[module.__name__] = {
                    'object': obj,
                    'priority': obj.priority,
            }

        # Load renderer
        for entry in pkg_resources.iter_entry_points("notification.renderer"):
            module = entry.load()
            self.log.debug("loading renderer module '%s'" % module.__name__)
            self.renderer[module.__name__] = {
                    'object': module(),
                    'priority': module.priority,
            }

        self.last_event = None

def load_tests(loader, tests, pattern):
    for entry in pkg_resources.iter_entry_points('qubes.tests.extra'):
        for test_case in entry.load()():
            tests.addTests(loader.loadTestsFromTestCase(test_case))

    try:
        qc = qubes.qubes.QubesVmCollection()
        qc.lock_db_for_reading()
        qc.load()
        qc.unlock_db()
        templates = [vm.name for vm in qc.values() if
                     isinstance(vm, qubes.qubes.QubesTemplateVm)]
    except OSError:
        templates = []

    for entry in pkg_resources.iter_entry_points(
            'qubes.tests.extra.for_template'):
        for test_case in entry.load()():
            for template in templates:
                tests.addTests(loader.loadTestsFromTestCase(
                    type(
                        '{}_{}_{}'.format(
                            entry.name, test_case.__name__, template),
                        (test_case,),
                        {'template': template}
                    )
                ))

    return tests

def test_group_chain(runner):

    # Attach a sub-group to a CLI and get execute it without arguments to make
    # sure both the sub-group and all the parent group's commands are present
    @good_cli.group()
    def sub_cli():
        """Sub CLI."""
        pass

    result = runner.invoke(good_cli)
    assert result.exit_code is 0
    assert sub_cli.name in result.output
    for ep in iter_entry_points('_test_click_plugins.test_plugins'):
        assert ep.name in result.output

    # Same as above but the sub-group has plugins
    @with_plugins(plugins=iter_entry_points('_test_click_plugins.test_plugins'))
    @good_cli.group()
    def sub_cli_plugins():
        """Sub CLI with plugins."""
        pass

    result = runner.invoke(good_cli, ['sub_cli_plugins'])
    assert result.exit_code is 0
    for ep in iter_entry_points('_test_click_plugins.test_plugins'):
        assert ep.name in result.output

    # Execute one of the sub-group's commands
    result = runner.invoke(good_cli, ['sub_cli_plugins', 'cmd1', 'something'])
    assert result.exit_code is 0
    assert result.output.strip() == 'passed'