def init():
"""Return top level command handler"""
@click.command()
@click.option('--cell', required=True,
envvar='TREADMILL_CELL',
callback=cli.handle_context_opt,
expose_value=False)
@click.option('--monitor', nargs=2, type=click.Tuple([str, int]),
multiple=True, required=True)
@click.option('--once', help='Run once.', is_flag=True, default=False)
@click.option('--interval', help='Wait interval between checks.',
default=_DEFAULT_INTERVAL)
@click.argument('name')
def controller(monitor, once, interval, name):
"""Control app monitors across cells"""
monitors = list(monitor)
while True:
intended_total = 0
actual_total = 0
intended = 0
for cellname, count in monitors:
if cellname == context.GLOBAL.cell:
intended = count
else:
actual = _count(cellname, name)
_LOGGER.info('state for cell %s, actual: %s, intended: %s',
cellname, actual, count)
intended_total += count
actual_total += actual
missing = intended_total - actual_total
# If there are missing instances, start them locally. If there are
# extra instances (missing < 0) just keep the indended state for
# the cell.
my_count = intended + max(0, missing)
_LOGGER.info('intended: %s, actual: %s, missing: %s, my count: %s',
intended_total, actual_total, missing, my_count)
_configure_monitor(name, my_count)
if once:
break
time.sleep(utils.to_seconds(interval))
return controller
python类Tuple()的实例源码
def __init__(self, name, app):
"""
Init command
:param name: Name of the command
:type name: str
:param app: The application
:type app: edmunds.application.Application
"""
super(TestCommand, self).__init__(name, app)
# Fetch nose options
config = Config(env=os.environ, files=all_config_files(), plugins=DefaultPluginManager())
nose_options = config.getParser(doc=TestProgram.usage()).option_list
# Override run-function to be able to load the click-options dynamically
# Dynamic click.option does not work for class-methods because of __click_params__
original_function = self.run
def run_wrapper(**kwargs):
return original_function(**kwargs)
self.run = run_wrapper
# Don't show --help
nose_options = filter(lambda nose_option: '--help' not in nose_option._long_opts, nose_options)
for nose_option in nose_options:
args = nose_option._short_opts + nose_option._long_opts
if not args:
continue
type_mapping = {
'string': str,
'int': int,
'long': int,
'float': float,
# 'complex': str,
# 'choice': str,
}
unsupported_attr = ['action', 'dest', 'const']
kwargs = {}
for attr in OptParseOption.ATTRS:
if attr in unsupported_attr:
continue
attr_value = getattr(nose_option, attr)
if attr_value is None:
continue
if attr == 'type':
attr_value = type_mapping[attr_value]
if nose_option.nargs > 1:
attr_value = click.Tuple([attr_value])
if attr == 'default':
if attr_value == optparse.NO_DEFAULT:
continue
kwargs[attr] = attr_value
click.option(*args[:2], **kwargs)(run_wrapper)