我们从Python开源项目中，提取了以下39个代码示例，用于说明如何使用click.Choice()。
def print_project_status(ctx): config = ctx.obj.config solution_num = ctx.obj.solution_num project_status = gather_project_status(ctx) # Print out a 'pretty' message showing project status, first up some project details click.secho("Project Details", bold=True) click.echo(" Project Name: " + config["project_name"]) click.echo(" Number of solutions generated: " + str(solution_num)) click.echo(" Latest Solution Folder: '" + config["project_name"] + "/solution" + str(solution_num) + "'") click.echo(" Language Choice: " + config["language"]) # And now details about what builds have been run/are passing. # This section uses lots (too many!) 'conditional expressions' to embed formatting into the output. click.secho("Build Status", bold=True) click.echo(" C Simulation: " + (click.style("Pass", fg='green') if "csim_pass" in project_status else (click.style("Fail", fg='red') if "csim_fail" in project_status else (click.style("Run (Can't get status)", fg='yellow') if "csim_done" in project_status else click.style("Not Run", fg='yellow'))))) click.echo(" C Synthesis: " + (click.style("Run", fg='green') if "syn_done" in project_status else click.style("Not Run", fg='yellow'))) click.echo(" Cosimulation: " + (click.style("Pass", fg='green') if "cosim_pass" in project_status else (click.style("Fail", fg='red') if "cosim_fail" in project_status else (click.style("Run (Can't get status)", fg='yellow') if "cosim_done" in project_status else click.style("Not Run", fg='yellow'))))) click.echo(" Export:" ) click.echo(" IP Catalog: " + (click.style("Run", fg='green') if "export_ip_done" in project_status else click.style("Not Run", fg='yellow'))) click.echo(" System Generator: " + (click.style("Run", fg='green') if "export_sysgen_done" in project_status else click.style("Not Run", fg='yellow'))) click.echo(" Export Evaluation: " + (click.style("Run", fg='green') if "evaluate_done" in project_status else click.style("Not Run", fg='yellow'))) ### Click Command Definitions ### # Report Command
def shell_complete_option(f): def callback(ctx, param, value): if not value or ctx.resilient_parsing: return if value == 'BASH': do_bash_complete() elif value == 'ZSH': do_zsh_complete() else: raise ValueError('Unsupported shell completion') click.get_current_context().exit(0) f = click.option('--shell-complete', cls=HiddenOption, is_eager=True, expose_value=False, type=click.Choice(SUPPORTED_SHELLS), callback=callback)(f) return f
def with_color(func): """Decorator which adds --color option if available.""" if pygments is None: def decorator(*args, **kwargs): kwargs["color"] = None return func(*args, **kwargs) else: decorator = click.option( "--color", default=None, type=click.Choice(["light", "dark"]), help=( "Colorize output. By default no color is used. " "Parameter means colorscheme of the terminal") )(func) decorator = six.wraps(func)(decorator) return decorator
def common_options(func): """Commonly used command options.""" def parse_preset(ctx, param, value): return PRESETS.get(value, (None, None)) def parse_private(ctx, param, value): return hex_from_b64(value) if value else None func = click.option('--private', default=None, help='Private.', callback=parse_private)(func) func = click.option( '--preset', default=None, help='Preset ID defining prime and generator pair.', type=click.Choice(PRESETS.keys()), callback=parse_preset )(func) return func
def codegen_options(f): f = click.option( "-c", "--categories", multiple=True, default=default_categories, type=click.Choice(all_categories), help="A list of the categories of inputs and outputs that should " "be enabled" )(f) f = click.option( "-f", "--param_file", type=click.File(), help="""YAML or JSON file describing the firmware module configuration to be flashed. This is the same file that is used for rosparam in the launch file.""" "code" )(f) f = click.option( "-p", "--plugin", multiple=True, help="Enable a specific plugin" )(f) f = click.option( "-t", "--target", help="PlatformIO target (e.g. upload)" )(f) f = click.option( "--status_update_interval", default=5, help="Minimum interval between driver status updates (in seconds)" )(f) return f
def main(): """pyFootball v.1 Early Alpha A python based football management simulator, very similar to and heavily based on Football Manager created by Sports Interactive. To start or load a game, run this file without any parameters. Coded by Justin Auger http://justnaugr.github.io\n Credits to Click for the great CLI. """ start = click.prompt('Would you like to start a new game or load a saved game?', type=click.Choice(['new','load'])) if start=='new': new_game() elif start=='load': load_game()
def raw_preset_option(help_action, callback): """ Get an option for OpenShift version presets. :param help_action: the helptext for the preset option :param callback: the callback for the preset option :return: the preset option """ return option( '--for', '-f', 'preset', type=Choice([ Preset.origin_master, Preset.ose_master, Preset.ose_32, Preset.ose_321, Preset.ose_33, ]), metavar='PRESET', expose_value=False, help=help_action + ' using a pre-set configuration for a specific version of OpenShift.', callback=callback, )
def option_simulation(f): def default_list(ctx, param, value): return list(value if len(value) > 0 else (2, )) options = [ click.option('-P', '--problem', type=click.Choice(['acoustic', 'tti']), help='Number of grid points along each axis'), click.option('-d', '--shape', default=(50, 50, 50), help='Number of grid points along each axis'), click.option('-s', '--spacing', default=(20., 20., 20.), help='Spacing between grid sizes in meters'), click.option('-n', '--nbpml', default=10, help='Number of PML layers'), click.option('-so', '--space-order', type=int, multiple=True, callback=default_list, help='Space order of the simulation'), click.option('-to', '--time-order', type=int, multiple=True, callback=default_list, help='Time order of the simulation'), click.option('-t', '--tn', default=250, help='End time of the simulation in ms'), ] for option in reversed(options): f = option(f) return f
def build_options(func): """Add "build" Click options to function. :param function func: The function to wrap. :return: The wrapped function. :rtype: function """ func = click.option('-a', '--banner-greatest-tag', is_flag=True, help='Override banner-main-ref to be the tag with the highest version number.')(func) func = click.option('-A', '--banner-recent-tag', is_flag=True, help='Override banner-main-ref to be the most recent committed tag.')(func) func = click.option('-b', '--show-banner', help='Show a warning banner.', is_flag=True)(func) func = click.option('-B', '--banner-main-ref', help="Don't show banner on this ref and point banner URLs to this ref. Default master.")(func) func = click.option('-i', '--invert', help='Invert/reverse order of versions.', is_flag=True)(func) func = click.option('-p', '--priority', type=click.Choice(('branches', 'tags')), help="Group these kinds of versions at the top (for themes that don't separate them).")(func) func = click.option('-r', '--root-ref', help='The branch/tag at the root of DESTINATION. Will also be in subdir. Default master.')(func) func = click.option('-s', '--sort', multiple=True, type=click.Choice(('semver', 'alpha', 'time')), help='Sort versions. Specify multiple times to sort equal values of one kind.')(func) func = click.option('-t', '--greatest-tag', is_flag=True, help='Override root-ref to be the tag with the highest version number.')(func) func = click.option('-T', '--recent-tag', is_flag=True, help='Override root-ref to be the most recent committed tag.')(func) func = click.option('-w', '--whitelist-branches', multiple=True, help='Whitelist branches that match the pattern. Can be specified more than once.')(func) func = click.option('-W', '--whitelist-tags', multiple=True, help='Whitelist tags that match the pattern. Can be specified more than once.')(func) return func
def set_type(kwargs, col_desc): if col_desc['type'] == 'string': pass elif col_desc['type'] == 'integer': kwargs["type"] = int elif col_desc['type'] == 'number': kwargs["type"] = float elif col_desc['type'] == 'boolean': kwargs["type"] = bool elif col_desc['type'] == 'enum': kwargs["type"] = click.Choice(col_desc['values']) else: raise Exception('Unknown column type %s' % col_desc['type'])
def option(flag, short_flag, type, help): """ Automagically detect whether a function has a default value for an option, and add an appropriately defaulted or required option decorator. This can only be used if you have wrapped the function in `clickutil.call.call`, since normally decorators strip out the information we need here. Note that if you want to provide default values for arguments only when calling from the command line, but not from python, then you must use `default_option` instead. For the sake of flexibility, we do not check the default value's type against the declared type. The `type` entry must be one of either: - a valid click type - a dict with a 'type' entry and an optional 'multiple' entry For example `type` could be: - `click.Choice(['choice_a', 'choice_b'])` - `{'multiple': True, 'type': str}` """ varname = flag.strip('-').replace('-', '_') def decorator(f): has_default, default = get_arg_default(f, varname) if has_default: return default_option(flag, short_flag, type, default, help)(f) else: return required_option(flag, short_flag, type, help)(f) return decorator
def __init__(self, config, **kwargs): """Base class to provide a command-based (similar to e.g. git) cli for freckles. This class parses the folders in the paths provided by the config element for so-called 'freckle adapters'. A freckle adapter is a collection of files that describe commandline-arguments and tasks lists to execute on a folder (more information: XXX) Args: config (FrecklesConfig): the config wrapper object kwargs (dict): additional arguments that are forwarded to the partent click.MultiCommand constructor """ click.MultiCommand.__init__(self, "freckles", result_callback=assemble_freckle_run, invoke_without_command=True, **kwargs) use_repo_option = click.Option(param_decls=["--use-repo", "-r"], required=False, multiple=True, help="extra context repos to use", is_eager=True, callback=download_extra_repos) output_option = click.Option(param_decls=["--output", "-o"], required=False, default="default", metavar="FORMAT", type=click.Choice(SUPPORTED_OUTPUT_FORMATS), is_eager=True, help="format of the output") no_run_option = click.Option(param_decls=["--no-run"], help='don\'t execute frecklecute, only prepare environment and print task list', type=bool, is_flag=True, default=False, required=False) version_option = click.Option(param_decls=["--version"], help='prints the version of freckles', type=bool, is_flag=True, is_eager=True, expose_value=False, callback=print_version) self.params = get_freckles_option_set() self.params.extend([ use_repo_option, output_option, no_run_option, version_option]) self.config = config self.profile_repo = ProfileRepo(self.config) # self.command_names.append(BREAK_COMMAND_NAME)
def __init__(self, current_command, config, **kwargs): """Base class to provide a command-based (similar to e.g. git) cli for frecklecute. This class parses the folders in the paths provided by the config element for so-called 'frecklecutables', (yaml) text files that contain a list of tasks and optionally command-line argument descriptions. More information: XXX Args: current_command (tuple): a tuple in the format (command_name, command_path), which is used for commands that are paths, instead of filenames in one of the known frecklecutable paths. Can be (None, None) if not a path. config (FrecklesConfig): the config wrapper object kwargs (dict): additional arguments that are forwarded to the partent click.MultiCommand constructor """ click.MultiCommand.__init__(self, "freckles", **kwargs) output_option = click.Option(param_decls=["--output", "-o"], required=False, default="default", metavar="FORMAT", type=click.Choice(SUPPORTED_OUTPUT_FORMATS), help="format of the output", is_eager=True) ask_become_pass_option = click.Option(param_decls=["--ask-become-pass", "-pw"], help='whether to force ask for a password, force ask not to, or let try freckles decide (which might not always work)', type=click.Choice(["auto", "true", "false"]), default="auto") version_option = click.Option(param_decls=["--version"], help='prints the version of freckles', type=bool, is_flag=True, is_eager=True, expose_value=False, callback=print_version) no_run_option = click.Option(param_decls=["--no-run"], help='don\'t execute frecklecute, only prepare environment and print task list', type=bool, is_flag=True, default=False, required=False) use_repo_option = click.Option(param_decls=["--use-repo", "-r"], required=False, multiple=True, help="extra context repos to use", is_eager=True, callback=download_extra_repos, expose_value=True) self.params = [use_repo_option, output_option, ask_become_pass_option, no_run_option, version_option] self.config = config # .trusted_repos = DEFAULT_FRECKLES_CONFIG.trusted_repos # local_paths = get_local_repos(trusted_repos, "frecklecutables") self.command_repo = CommandRepo(config=self.config, additional_commands=[current_command]) self.current_command = current_command[0]
def distrib_option(**attrs): attrs.setdefault("help", "Value distribution") return str_option("--distrib", type=click.Choice([m.name for m in ValueDistribution]), default="normal", **attrs)
def mask_option(**attrs): attrs.setdefault("help", "Value distribution: Hot pixel mask") return str_option("--mask", type=click.Choice([m.name for m in ValueMask]), **attrs)
def point_distrib_option(**attrs): attrs.setdefault("help", "Voronoi/DLA: Point cloud distribution") return str_option("--point-distrib", type=click.Choice([m.name for m in PointDistribution]), default="random", **attrs)
def saturation_distrib_option(**attrs): attrs.setdefault("help", "HSV: Override value distribution for saturation") return str_option("--saturation-distrib", type=click.Choice([m.name for m in ValueDistribution]), default=None, **attrs)
def brightness_distrib_option(**attrs): attrs.setdefault("help", "HSV: Override value distribution for brightness") return str_option("--brightness-distrib", type=click.Choice([m.name for m in ValueDistribution]), default=None, **attrs)
def screenshots_options(func): return click.option('-o', '--orientation', type=click.Choice(ORIENTATIONS), help='Screen orientation')( click.option('-m', '--mac-res', type=click.Choice(MAC_RESOLUTIONS), help='Screen resolution on OS X')( click.option('-w', '--win-res', type=click.Choice(WIN_RESOLUTIONS), help='Screen resolution on Windows')( click.option('-q', '--quality', type=click.Choice(QUALITIES), help='Quality of the screenshot')( click.option('-l', '--local', type=click.Choice(LOCALS), help='Local testing')( click.option('-t', '--wait-time', type=click.Choice(WAIT_TIMES), help='Seconds before taking screenshot')( click.option('-c', '--callback-url', help='Results will be send to this URL')(func)))))))
def convert_primitive(ast): import click mapping = { 'Int': int, 'Str': str, 'Float': float, 'Color': str, 'Bool': bool } # TODO: it would be a good idea to refactor this someday, but until then # just handle the few predicates we know about. predicate = ast['predicate'] if predicate: if predicate['name'] == 'Choices' and ast['name'] == 'Str': return click.Choice(predicate['choices']) elif predicate['name'] == 'Range' and ast['name'] == 'Int': start = predicate['start'] end = predicate['end'] # click.IntRange is always inclusive if start is not None and not predicate['inclusive-start']: start += 1 if end is not None and not predicate['inclusive-end']: end -= 1 return click.IntRange(start, end) elif predicate['name'] == 'Range' and ast['name'] == 'Float': # click.FloatRange will be in click 7.0, so for now the # range handling will just fallback to qiime2. return mapping['Float'] else: raise NotImplementedError() else: return mapping[ast['name']]
def init(): """Return top level command handler.""" @click.command() @click.option('--run/--no-run', is_flag=True, default=False) @click.option('--master-id', required=True, type=click.Choice(['1', '2', '3'])) @click.option('--ldap-pwd', help='LDAP password (clear text of path to file).') @click.pass_context def master(ctx, run, master_id, ldap_pwd): """Installs Treadmill master.""" ctx.obj['PARAMS']['zookeeper'] = context.GLOBAL.zk.url ctx.obj['PARAMS']['ldap'] = context.GLOBAL.ldap.url ctx.obj['PARAMS']['master_id'] = master_id dst_dir = ctx.obj['PARAMS']['dir'] profile = ctx.obj['PARAMS'].get('profile') if ldap_pwd: ctx.obj['PARAMS']['ldap_pwd'] = ldap_pwd for master in ctx.obj['PARAMS']['masters']: # pylint: disable=E1136 if int(master['idx']) == int(master_id): ctx.obj['PARAMS'].update({'me': master}) run_sh = None if run: run_sh = os.path.join(dst_dir, 'treadmill', 'bin', 'run.sh') bootstrap.install( 'master', dst_dir, ctx.obj['PARAMS'], run=run_sh, profile=profile, ) return master
def init(): """Return top level command handler.""" @click.command() @click.option('--run/--no-run', is_flag=True, default=False) @click.option('--master-id', required=True, type=click.Choice(['1', '2', '3'])) @click.pass_context def zookeeper(ctx, run, master_id): """Installs Treadmill master.""" ctx.obj['PARAMS']['zookeeper'] = context.GLOBAL.zk.url ctx.obj['PARAMS']['ldap'] = context.GLOBAL.ldap.url ctx.obj['PARAMS']['master_id'] = master_id dst_dir = ctx.obj['PARAMS']['dir'] profile = ctx.obj['PARAMS'].get('profile') for master in ctx.obj['PARAMS']['masters']: # pylint: disable=E1136 if int(master['idx']) == int(master_id): ctx.obj['PARAMS'].update({'me': master}) run_sh = None if run: run_sh = os.path.join(dst_dir, 'treadmill', 'bin', 'run.sh') bootstrap.install( 'zookeeper', dst_dir, ctx.obj['PARAMS'], run=run_sh, profile=profile, ) return zookeeper
def output_options(f): f = click.option('--output-format', '-f', help='output format', type=click.Choice(outfmt.available_formats), default='txt', show_default=True)(f) f = click.option('--output', '-o', help='output file', default='-')(f) return f
def console(ctx, container: str, user: str, tty: bool): """See command Help""" if len(ctx.obj['CTS']) is not 0: ct_choice = click.Choice(ctx.obj['CTS']) ct_choice.convert(container, None, ctx) ctx.obj['STAKKR'].console(container, _get_cmd_user(user, container), tty)
def exec_cmd(ctx, user: str, container: str, command: tuple, tty: bool): """See command Help""" if len(ctx.obj['CTS']) is not 0: click.Choice(ctx.obj['CTS']).convert(container, None, ctx) ctx.obj['STAKKR'].exec_cmd(container, _get_cmd_user(user, container), command, tty)
def choose_property_files(): return click.Choice([p.name for p in list(TransmartBatch().get_property_files())])
def trait_to_option(name, trait): if is_trait(trait): if isinstance(trait, Int): return click.Option(param_decls=('--' + name,), default=trait.default_value, type=click.IntRange(trait.min, trait.max)) elif isinstance(trait, Float): return click.Option(param_decls=('--' + name,), default=trait.default_value, type=float) elif isinstance(trait, Complex): return click.Option(param_decls=('--' + name,), default=trait.default_value, type=complex) elif isinstance(trait, Bool): return click.Option(param_decls=('--' + name,), default=trait.default_value, is_flag=True) elif isinstance(trait, Unicode): return click.Option(param_decls=('--' + name,), default=trait.default_value, type=str) elif isinstance(trait, Enum): # FIXME: trait.values should be strings return click.Option(param_decls=('--' + name,), default=str(trait.default_value), type=click.Choice(list(map(str, trait.values)))) elif isinstance(trait, Tuple): return click.Option(param_decls=('--' + name,), default=trait.default_value, type=tuple( trait_to_type(t) for t in trait._traits)) else: raise InvalidValue('Trait conversion is not supported for: ' '{}'.format(trait)) else: raise InvalidType('Trait should be instance of {}'.format(TraitType))
def main(user, flag): global _user global _league _user = user _league = user.get_league() if flag: click.clear() print('%s Managerial office' % _user.get_team().get_name()) print('-'.ljust(39,'-')) nav = click.prompt('What would you like to do?', type=click.Choice(['help','inbox','schedule','squad','standings','personal', 'save', 'exit'])) if nav == 'help': print('\nWhile in your office, you have a variety of resources available to you. You may:\n\n' 'inbox : access any new mail you\'ve received, whether they be newsletters, injury news, player communications, or transfer offers\n' 'schedule : take a look at upcoming games and past results of both your team and other teams in the league\n' 'squad : check on how your players are doing, including their stats based on recent training sessions\n' 'standings : see how your team ranks up on the table, along with other useful information and stats\n' 'personal : see your own personal stats and information\n' 'save : save your in-game progress\n' 'exit : exit out of the game, although why would you do that?\n') main(_user, False) elif nav == 'exit': pass elif nav == 'inbox': inbox() elif nav == 'schedule': schedule() elif nav == 'squad': squad() elif nav == 'standings': standings() elif nav == 'personal': personal() elif nav == 'save': save()
def repository_argument(func): """ Add the repository selection argument to the decorated command func. :param func: Click CLI command to decorate :return: decorated CLI command """ return argument( 'repository', nargs=1, type=Choice([ Repository.origin, Repository.enterprise, Repository.web_console, Repository.web_console_server, Repository.source_to_image, Repository.metrics, Repository.logging, Repository.online, Repository.online_hibernation, Repository.release, Repository.aoscdjobs, Repository.openshift_ansible, Repository.jenkins, Repository.wildfly, Repository.jenkins_plugin, Repository.jenkins_sync_plugin, Repository.jenkins_client_plugin, Repository.jenkins_login_plugin, Repository.image_registry, Repository.cluster_operator, Repository.kubernetes_metrics_server, ]) )(func)
def package_options(func): """ Add all of the package options to the decorated command func. :param func: Click CLI command to decorate :return: decorated CLI command """ return option( '--stage', '-s', 'upgrade_stage', type=Choice(['current', 'next', 'fork', 'base', 'crio']), help='Update the current stage, upgrade to next default stage, or choose a stage', )(func)
def __init__(self, option): if option.choices: type_ = click.Choice(option.choices) elif option.min is not None or option.max is not None: type_ = click.IntRange(min=option.min, max=option.max) else: type_ = TYPES[option.type_name] super().__init__( param_decls=option.cli, help=option.help, default=None, type=type_) self.metadata = option.metadata
def test_string_with_choices_type(self): class Option(StringOption): metadata = 'name' choices = ['one', 'two', 'three'] cli = ['--name'] click_option = ClickObjectOption(Option) self.assertIsInstance(click_option.type, click.Choice) self.assertEqual(click_option.type.choices, ['one', 'two', 'three'])
def _add_format_option(self, command, action): """ Add format option. This option defines the format type of the command output. This option is attached only on `ListCommand` and `RetrieveCommand`. :param command: Command instance. :param action: Action type, e.g. 'list', 'retrieve', etc. """ if action in self.READ_ACTIONS: return click.option( '--format', type=click.Choice(['json', 'table']), default='json')(command)
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('--once', help='Run once.', is_flag=True, default=False) @click.option('--interval', help='Wait interval between checks.', default=_DEFAULT_INTERVAL) @click.option('--threshold', help='Number of failed checks before reap.', default=1) @click.option('--proto', help='Endpoint protocol.', default='tcp', type=click.Choice(['tcp', 'udp'])) @click.argument('pattern') @click.argument('endpoint') @click.argument('command', nargs=-1) def reaper(once, interval, threshold, proto, pattern, endpoint, command): """Removes unhealthy instances of the app. The health check script reads from STDIN and prints to STDOUT. The input it list of instance host:port, similar to discovery. Output - list of instances that did not pass health check. For example, specifying awk '{print $1}' as COMMAND will remove all instances. """ command = list(command) failed = collections.Counter() while True: failed.update(_health_check(pattern, proto, endpoint, command)) for instance, count in failed.items(): _LOGGER.info('Failed: %s, count: %s', instance, count) reaped = _reap([instance for instance, count in failed.items() if count > threshold]) for instance in reaped: del failed[instance] if once: break time.sleep(utils.to_seconds(interval)) return reaper
def explain_group(parent): """Scheduler explain CLI group.""" def _print_frame(df): """Prints dataframe.""" if not df.empty: pd.set_option('display.max_rows', None) pd.set_option('float_format', lambda f: '%f' % f) pd.set_option('expand_frame_repr', False) print(df.to_string(index=False)) @parent.group() def explain(): """Explain scheduler internals""" pass @explain.command() @click.option('--instance', help='Application instance') @click.option('--partition', help='Cell partition', default='_default') @cli.admin.ON_EXCEPTIONS def queue(instance, partition): """Explain the application queue""" cell_master = make_readonly_master() frame = reports.explain_queue(cell_master.cell, partition, pattern=instance) _print_frame(frame) @explain.command() @click.argument('instance') @click.option('--mode', help='Tree traversal method', type=click.Choice(reports.WALKS.keys()), default='default') @cli.admin.ON_EXCEPTIONS def placement(instance, mode): """Explain application placement""" cell_master = make_readonly_master() if instance not in cell_master.cell.apps: cli.bad_exit('Instance not found.') app = cell_master.cell.apps[instance] if app.server: cli.bad_exit('Instace already placed on %s' % app.server) frame = reports.explain_placement(cell_master.cell, app, mode) _print_frame(frame) del queue del placement
def init(): """Top level command handler.""" @click.command() @click.option('-r', '--register', required=False, default=False, is_flag=True, help='Register as /nodeinfo in Zookeeper.') @click.option('-p', '--port', required=False, default=0) @click.option('-a', '--auth', type=click.Choice(['spnego'])) @click.option('-m', '--modules', help='API modules to load.', required=False, type=cli.LIST) @click.option('-t', '--title', help='API Doc Title', default='Treadmill Nodeinfo REST API') @click.option('-c', '--cors-origin', help='CORS origin REGEX') def server(register, port, auth, modules, title, cors_origin): """Runs nodeinfo server.""" if port == 0: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.bind(('0.0.0.0', 0)) port = sock.getsockname()[1] sock.close() hostname = sysinfo.hostname() hostport = '%s:%s' % (hostname, port) if register: zkclient = context.GLOBAL.zk.conn zkclient.add_listener(zkutils.exit_on_lost) appname = 'root.%s#%010d' % (hostname, os.getpid()) path = z.path.endpoint(appname, 'tcp', 'nodeinfo') _LOGGER.info('register endpoint: %s %s', path, hostport) zkutils.create(zkclient, path, hostport, acl=[_SERVERS_ACL], ephemeral=True) _LOGGER.info('Starting nodeinfo server on port: %s', port) utils.drop_privileges() api_paths = [] if modules: api_paths = api.init(modules, title.replace('_', ' '), cors_origin) rest_server = rest.TcpRestServer(port, auth_type=auth, protect=api_paths) rest_server.run() return server
def init(): """App main.""" # pylint: disable=W0612 @click.command() @click.option('-p', '--port', help='Port for TCP server') @click.option('-s', '--socket', help='Socket for UDS server') @click.option('-a', '--auth', type=click.Choice(['spnego'])) @click.option('-t', '--title', help='API Doc Title', default='Treadmill REST API') @click.option('-c', '--cors-origin', help='CORS origin REGEX', required=True) @click.option('--workers', help='Number of workers', default=5) @click.option('--interval', help='interval to refresh cgroups', default=60) def server(port, socket, auth, title, cors_origin, workers, interval): """Create pge server to provide authorize service.""" (base_api, cors) = api.base_api(title, cors_origin) endpoint = cgroup_api.init(base_api, cors, interval=interval) if not endpoint.startswith('/'): endpoint = '/' + endpoint if port: rest_server = rest.TcpRestServer( port, auth_type=auth, protect=[endpoint], workers=workers ) # TODO: need to rename that - conflicts with import socket. elif socket: rest_server = rest.UdsRestServer(socket) else: click.echo('port or socket must be specified') sys.exit(1) try: rest_server.run() except sock.error as sock_err: print(sock_err) if sock_err.errno == errno.EADDRINUSE: # TODO: hack, but please keep it for now, otherwise on the # setup several master processes run on same server # lookup api (listen on port 8080) is in tight loop. time.sleep(5) return server
def load_dump(collection_dump, collection_name): """ Restore an index from a dump file. :param collection_dump: Path to a local gzipped dump to load. :param collection_name: Name for the local index to restore the dump to. Optional; will be derived from the dump name, at your own risk. Note that the pipeline will add a "owa_" prefix string to the collection name, to ensure the proper mapping and settings are applied. """ available_dumps = glob(os.path.join(LOCAL_DUMPS_DIR, '*/*.gz')) if not collection_dump: choices = [] for i, dump in enumerate(available_dumps): choices.append(unicode(i+1)) click.secho('{i}) {dump}'.format(i=i+1, dump=dump), fg='green') dump_idx = click.prompt('Choose one of the dumps listed above', type=click.Choice(choices)) collection_dump = available_dumps[int(dump_idx) - 1] collection = os.path.abspath(collection_dump) collection_id = '_'.join(collection.split('/')[-1].split('.')[0].split('_')[:2]) if not collection_name: collection_name = collection_id.replace('owa_', '') source_definition = { 'id': collection_id, 'extractor': 'ocd_backend.extractors.staticfile.StaticJSONDumpExtractor', 'transformer': 'ocd_backend.transformers.BaseTransformer', 'loader': 'ocd_backend.loaders.ElasticsearchLoader', 'item': 'ocd_backend.items.LocalDumpItem', 'dump_path': collection, 'index_name': collection_name } click.secho(str(source_definition), fg='yellow') setup_pipeline(source_definition) click.secho('Queued items from {}. Please make sure your Celery workers' ' are running, so the loaded items are processed.'.format(collection), fg='green') # Register commands explicitly with groups, so we can easily use the docstring # wrapper
def option_performance(f): """Defines options for all aspects of performance tuning""" _preset = { # Fixed 'O1': {'autotune': True, 'dse': 'basic', 'dle': 'basic'}, 'O2': {'autotune': True, 'dse': 'advanced', 'dle': 'advanced'}, 'O3': {'autotune': True, 'dse': 'aggressive', 'dle': 'advanced'}, # Parametric 'dse': {'autotune': True, 'dse': ['basic', 'advanced', 'speculative', 'aggressive'], 'dle': 'advanced'}, 'dle': {'autotune': True, 'dse': 'advanced', 'dle': ['basic', 'advanced']} } def from_preset(ctx, param, value): """Set all performance options according to bench-mode preset""" ctx.params.update(_preset[value]) return value def from_value(ctx, param, value): """Prefer preset values and warn for competing values.""" return ctx.params[param.name] or value options = [ click.option('-bm', '--bench-mode', is_eager=True, callback=from_preset, expose_value=False, default='O2', type=click.Choice(['O1', 'O2', 'O3', 'dse', 'dle']), help='Choose what to benchmark; ignored if execmode=run'), click.option('--arch', default='unknown', help='Architecture on which the simulation is/was run'), click.option('--dse', callback=from_value, type=click.Choice(['noop'] + configuration._accepted['dse']), help='Devito symbolic engine (DSE) mode'), click.option('--dle', callback=from_value, type=click.Choice(['noop'] + configuration._accepted['dle']), help='Devito loop engine (DLE) mode'), click.option('-a', '--autotune', is_flag=True, callback=from_value, help='Switch auto tuning on/off'), ] for option in reversed(options): f = option(f) return f
