Python click 模块，Abort() 实例源码

def create_module_workingcopy_template(module_name, module_version, body):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.post('/rest/rest/modules' + module_name + '/' + module_version + '/workingcopy', body=file_body)
    utils.pretty_print(response)

def get_module(module_name, module_version, module_type):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_type and module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    if module_version is None and module_type is None:
        response = client.get('/rest/modules/' + module_name)
        utils.pretty_print(response)
    if module_version and module_type is None:
        response = client.get('/rest/modules/' + module_name + '/' + module_version)
        utils.pretty_print(response)
    if module_version and module_type:
        response = client.get('/rest/modules/' + module_name + '/' + module_version + '/' + module_type)
        utils.pretty_print(response)

def update_module_workingcopy_template(module_name, module_version, body):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.put('/rest/modules/' + module_name + '/' + module_version + '/workingcopy/templates',
                          body=file_body)
    utils.pretty_print(response)

def get_properties(application_name, platform_name, path, timestamp):
    params = {}
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    if path is None:
        print('--path required')
        raise click.Abort()
    params['path'] = path
    if timestamp:
        params['timestamp'] = "timestamp"
    client = Client()
    response = client.get(
        '/rest/applications/' + application_name + '/platforms/' + platform_name + '/properties', params=params)
    utils.pretty_print(response)

def take_application_platform_snapshot(application_name, platform_name, body):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.post(
        '/rest/applications/' + application_name + '/platforms/' + platform_name + '/take_snapshot', body=file_body)
    utils.pretty_print(response)

def create_application_platform(application_name, from_application, from_platform, body):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if from_application is None and from_platform:
        print('--from_application required when --from_platform is given')
        raise click.Abort()
    if from_application and from_platform is None:
        print('--from_platform required when --from_application is given')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.post('/rest/applications/' + application_name + '/platforms', body=file_body)
    utils.pretty_print(response)

def update_application_platform(application_name, copy_properties_for_upgraded_modules, body):
    params = {}
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    if copy_properties_for_upgraded_modules:
        params['copy_properties_for_upgraded_modules'] = copy_properties_for_upgraded_modules
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.put('/rest/applications/' + application_name + '/platforms', params=params, body=file_body)
    utils.pretty_print(response)

def create_template_package_workingcopy(package_name, package_version, template_name, body):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    if template_name is None:
        print('--template_name required')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.post(
        '/rest/templates/packages/' + package_name + '/' + package_version + '/workingcopy/templates/' + template_name,
        body=file_body)
    utils.pretty_print(response)

def update_template_package_workingcopy(package_name, package_version, template_name, body):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    if template_name is None:
        print('--template_name required')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.put(
        '/rest/templates/packages/' + package_name + '/' + package_version + '/workingcopy/templates/' + template_name,
        body=file_body)
    utils.pretty_print(response)

def do_start_build_stuff(ctx):
    config = ctx.obj.config
    solution_num = ctx.obj.solution_num
    try:
        file = click.open_file("run_hls.tcl","w")
        file.write("open_project " + config["project_name"] + "\n")
        file.write("set_top " + config["top_level_function_name"] + "\n")
        for src_file in config["src_files"]:
            file.write("add_files " + config["src_dir_name"] + "/" + src_file + "\n")
        for tb_file in config["tb_files"]:
            file.write("add_files -tb " + config["tb_dir_name"] + "/" + tb_file + "\n")
        if ctx.params['keep']:
            file.write("open_solution -reset \"solution" + str(solution_num) + "\"" + "\n")
        else:
            file.write("open_solution \"solution" + str(solution_num) + "\"" + "\n")
        file.write("set_part \{" + config["part_name"] + "\}" + "\n")
        file.write("create_clock -period " + config["clock_period"] + " -name default" + "\n")
        return file
    except OSError:
        click.echo("Woah! Couldn't create a Tcl run file in the current folder!")
        raise click.Abort()

# Function to write a default build into the HLS Tcl build script.

def build_end_callback(ctx,sub_command_returns,keep,report):
    # Catch the case where no subcommands have been issued and offer a default build
    if not sub_command_returns:
        if click.confirm("No build stages specified, would you like to run a default sequence using all the build stages?", abort=True):
            do_default_build(ctx)
    ctx.obj.file.write("exit" + "\n")
    ctx.obj.file.close()
    # Call the Vivado HLS process
    hls_processs = subprocess.run(["vivado_hls", "-f", "run_hls.tcl"])
    # Check return status of the HLS process.
    if hls_processs.returncode < 0:
        raise click.Abort()
    elif hls_processs.returncode > 0:
        click.echo("Warning: HLS Process returned an error, skipping report opening!")
        raise click.Abort()
    else:
        do_end_build_stuff(ctx,sub_command_returns,report)

# csim subcommand

def cli(ctx):
    """Helper tool for using Vivado HLS through the command line. If no arguments are specified then a default run is executed which includes C simulation, C synthesis, Cosimulation and export for both Vivado IP Catalog and System Generator. If any of the run options are specified then only those specified are performed."""
    # Generate a default config dict and then load in the local config file.
    config = generate_default_config();
    config_loaded = get_vars_from_file('hls_config.py')
    errors = []
    parse_config_vars(config_loaded, config, errors)
    if len(errors) != 0:
        for err in errors:
            print(err)
        print("Config Errors, exiting...")
        raise click.Abort()
    # Store the loaded config in an object within the Click context so it is available to all commands.
    obj = hlsclt_internal_object(config)
    ctx.obj = obj
    pass

# Add Click Sub Commands

def check_name(ctx, param, value):
    """
        Checks the project name to be a valid name
    :param ctx: app context
    :param param: parameter
    :param value: the parameter value
    :return:
    """
    regex = re.compile(r'^[a-zA-Z_]*')
    if regex.match(value):
        return value
    while True:
        try:
            click.echo(
                click.style("{}".format(value), bold=True) +
                ' is not a valid python package name. try something else',
                err=True)
            value = param.prompt_for_value(ctx)
            if regex.match(value):
                return value
            continue
        except (KeyboardInterrupt, EOFError):
            raise ctx.Abort()

def cli(*args, **kwargs):
    """
    CSVtoTable commandline utility.
    """
    # Convert CSV file
    content = convert.convert(kwargs["input_file"], **kwargs)

    # Serve the temporary file in browser.
    if kwargs["serve"]:
        convert.serve(content)
    # Write to output file
    elif kwargs["output_file"]:
        # Check if file can be overwrite
        if (not kwargs["overwrite"] and
                not prompt_overwrite(kwargs["output_file"])):
            raise click.Abort()

        convert.save(kwargs["output_file"], content)
        click.secho("File converted successfully: {}".format(
            kwargs["output_file"]), fg="green")
    else:
        # If its not server and output file is missing then raise error
        raise click.BadOptionUsage("Missing argument \"output_file\".")

def generate_sdk(ctx, sdk_type, stage, outdir):
    # type: (click.Context, str, str, str) -> None
    factory = ctx.obj['factory']  # type: CLIFactory
    config = factory.create_config_obj(stage)
    session = factory.create_botocore_session()
    client = TypedAWSClient(session)
    deployed = config.deployed_resources(stage)
    if deployed is None:
        click.echo("Could not find API ID, has this application "
                   "been deployed?", err=True)
        raise click.Abort()
    else:
        rest_api_id = deployed.rest_api_id
        api_gateway_stage = deployed.api_gateway_stage
        client.download_sdk(rest_api_id, outdir,
                            api_gateway_stage=api_gateway_stage,
                            sdk_type=sdk_type)

def wait_for_deployment(config, application, version, release, timeout, interval):
    '''Wait for all pods to become ready'''
    namespace = config.get('kubernetes_namespace')
    kubectl_login(config)
    deployment_name = '{}-{}-{}'.format(application, version, release)
    cutoff = time.time() + timeout
    while time.time() < cutoff:
        data = kubectl_get(namespace, 'pods', '-l',
                           'application={},version={},release={}'.format(application, version, release))
        pods = data['items']
        pods_ready = 0
        for pod in pods:
            if pod['status'].get('phase') == 'Running':
                all_containers_ready = True
                for cont in pod['status'].get('containerStatuses', []):
                    if not cont.get('ready'):
                        all_containers_ready = False
                if all_containers_ready:
                    pods_ready += 1
        if pods and pods_ready >= len(pods):
            return
        info('Waiting up to {:.0f} more secs for deployment '
             '{} ({}/{} pods ready)..'.format(cutoff - time.time(), deployment_name, pods_ready, len(pods)))
        time.sleep(interval)
    raise click.Abort()

def run():
    global creatorQueue, tosQueue, verifierQueue, logQueue, requestSleepTimer, requestSleepTimerB
    creatorQueue = Queue.Queue()
    tosQueue = Queue.Queue()
    verifierQueue = Queue.Queue()
    logQueue = Queue.Queue()
    requestSleepTimer  = 0.1
    requestSleepTimerB = 0.1

    try:
        main(standalone_mode=False)
    except (EOFError, KeyboardInterrupt):
        raise click.Abort()
    except click.ClickException as e:
        e.show()
        sys.exit(e.exit_code)
    except click.Abort:
        global accountStore
        accountStore.done()
        click.echo('Aborted!', file=sys.stderr)
        sys.exit(1)

    sys.exit(0)

def init(ctx, project_name):
    if os.path.isdir(project_name):
        click.echo('Already a directory named: %s' % project_name)
        raise click.Abort()
    pipetree_dir = os.path.join(project_name, '.pipetree')
    config = os.path.join(pipetree_dir, 'config.json')
    pipeline_config = os.path.join(project_name, 'pipeline.json')
    os.makedirs(pipetree_dir)
    with open(config, 'w') as f:
        f.write(DEFAULT_CONFIG % project_name)
    with open(pipeline_config, 'w') as f:
        f.write(DEFAULT_PIPELINE_CONFIG % project_name)
    lib_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    cat_src_dir = os.path.join(lib_dir, 'examples', 'cats')
    cat_dst_dir = os.path.join(project_name, 'cat_imgs')
    shutil.copytree(cat_src_dir, cat_dst_dir)

    py_package = os.path.join(project_name, project_name)
    os.makedirs(py_package)
    with open(os.path.join(py_package, '__init__.py'), 'w'):
        pass
    with open(os.path.join(py_package, 'main.py'), 'w') as f:
        f.write(DEFAULT_HANDLERS)
    click.echo("Created new project %s" % project_name)

def archive(ctx):

    lSessionId = 'archive'

    env = ctx.obj

    ensureVivado(env)

    lOpenCmds = [
        'open_project %s' % join(env.projectPath, 'top', 'top'),
    ]
    lArchiveCmds = [
        'archive_project %s -force' % join(env.projectPath, '{}.xpr.zip'.format(env.projectConfig['name'])),
    ]

    from ..tools.xilinx import VivadoOpen, VivadoConsoleError
    try:
        with VivadoOpen(lSessionId) as lTarget:
            lTarget(lOpenCmds)
            lTarget(lArchiveCmds)
    except VivadoConsoleError as lExc:
        secho("Vivado errors detected\n" +
              "\n".join(lExc.errors), fg='red')
        raise click.Abort()

def create_module(from_module_name, from_module_version, from_is_working_copy, body):
    params = {}
    if from_module_name is None and from_module_version:
        print('--from_module_name required when --from_module_version is given')
        raise click.Abort()
    if from_module_name and from_module_version is None:
        print('--from_module_version required when --from_module_name is given')
        raise click.Abort()
    if body is None:
        print('--body required')
        raise click.Abort()
    if from_module_name:
        params['from_module_name'] = from_module_name
    if from_module_version:
        params['from_module_version'] = from_module_version
    if from_is_working_copy:
        params['from_is_working_copy'] = from_is_working_copy
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.post('/rest/modules', params=params, body=file_body)
    utils.pretty_print(response)

def create_module_release(module_name, module_version, release_version):
    params = {}
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    params['module_name'] = module_name
    params['release_version'] = release_version
    client = Client()
    response = client.post('/rest/modules/create_release', params=params)
    utils.pretty_print(response)

def delete_module_release(module_name, module_version):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.delete('/rest/rest/modules' + module_name + '/' + module_version + '/release')
    utils.pretty_print(response)

def delete_module_workingcopy(module_name, module_version):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.delete('/rest/rest/modules' + module_name + '/' + module_version + '/workingcopy')
    utils.pretty_print(response)

def get_module_release(module_name, module_version):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/modules/' + module_name + '/' + module_version + '/release')
    utils.pretty_print(response)

def get_module_release_template(module_name, module_version, template_name):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/modules/' + module_name + '/' + module_version + '/release/templates' + template_name)
    utils.pretty_print(response)

def get_module_release_templates(module_name, module_version):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/modules/' + module_name + '/' + module_version + '/release/templates')
    utils.pretty_print(response)

def get_module_workingcopy(module_name, module_version):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/modules/' + module_name + '/' + module_version + '/workingcopy')
    utils.pretty_print(response)

def get_module_workingcopy_templates(module_name, module_version):
    if module_name is None:
        print('--module_name required')
        raise click.Abort()
    if module_version is None:
        print('--module_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/modules/' + module_name + '/' + module_version + '/workingcopy/templates/')
    utils.pretty_print(response)

def perform_search_modules(terms):
    params = {}
    if terms is None:
        print('--terms required')
        raise click.Abort()
    params['terms'] = terms
    client = Client()
    response = client.post('/rest/modules/perform_search', params=params)
    utils.pretty_print(response)

def search_module(terms):
    params = {}
    if terms is None:
        print('--terms required')
        raise click.Abort()
    params['temrs'] = terms
    client = Client()
    response = client.post('/rest/modules/search', params=params)
    utils.pretty_print(response)

def update_module(body):
    if body is None:
        print('--body required')
        raise click.Abort()
    file = open(body, "r")
    file_body = file.read()
    file.close()
    client = Client()
    response = client.put('/rest/modules/', body=file_body)
    utils.pretty_print(response)

def delete_profile(profile_name):
    if profile_name is None:
        print('--profile_name required')
        raise click.Abort()
    config_writer = ConfigFileWriter()
    config_writer.remove_config_section(profile_name)
    reader = ConfigFileReader()
    if reader.get_profile() == profile_name:
        config_writer.remove_profile()
    credentials_writer = ConfigFileWriter()
    credentials_writer.remove_credentials_section(profile_name)

def set_profile(profile_name):
    if profile_name is None:
        print('--profile_name required')
        raise click.Abort()
    config_writer = ConfigFileWriter()
    config = {'profile': profile_name}
    config_writer.update_config('config', config, True)

def get_global_properties(application_name, platform_name, timestamp):
    params = {}
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    params['path'] = "#"
    if timestamp:
        params['timestamp'] = "timestamp"
    client = Client()
    response = client.get('/rest/applications/' + application_name + '/platforms/' + platform_name + '/properties',
                          params=params)
    utils.pretty_print(response)

def get_global_properties_usage(application_name, platform_name):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    client = Client()
    response = client.get(
        '/rest/applications/' + application_name + '/platforms/' + platform_name + '/global_properties_usage')
    utils.pretty_print(response)

def get_application_platform_snapshots(application_name, platform_name):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/applications/' + application_name + '/platforms/' + platform_name + '/snapshots')
    utils.pretty_print(response)

def restore_application_platform_snapshots(application_name, platform_name):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    client = Client()
    response = client.post(
        '/rest/applications/' + application_name + '/platforms/' + platform_name + '/restore_snapshot')
    utils.pretty_print(response)

def delete_application_platform(application_name, platform_name):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    if platform_name is None:
        print('--platform_name required')
        raise click.Abort()
    client = Client()
    response = client.delete('/rest/applications/' + application_name + '/platforms/' + platform_name)
    utils.pretty_print(response)

def perform_search_application_platforms(application_name, platform_name):
    params = {}
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    params['application_name'] = application_name
    if platform_name:
        params['platform_name'] = platform_name
    client = Client()
    response = client.post('/rest/applications/platforms/perform_search', params=params)
    utils.pretty_print(response)

def get_application(application_name):
    if application_name is None:
        print('--application_name required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/applications/' + application_name)
    utils.pretty_print(response)

def get_application_from_module(module, version, type):
    if module is None:
        print('--module required')
        raise click.Abort()
    if version is None:
        print('--version required')
        raise click.Abort()
    if type is None:
        print('--type required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/applications/using_module/' + module + '/' + version + '/' + type)
    utils.pretty_print(response)

def perform_search_applications(name):
    params = {}
    if name is None:
        print('--name required')
        raise click.Abort()
    else:
        params['name'] = name
    client = Client()
    response = client.post('/rest/applications/perform_search', params=params)
    utils.pretty_print(response)

def delete_template_package_release(package_name, package_version):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    client = Client()
    response = client.delete('/rest/templates/packages/' + package_name + '/' + package_version + '/release')
    utils.pretty_print(response)

def delete_template_package_workingcopy(package_name, package_version):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    client = Client()
    response = client.delete('/rest/templates/packages/' + package_name + '/' + package_version + '/workingcopy')
    utils.pretty_print(response)

def get_template_package_release(package_name, package_version, template_name):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    if template_name is None:
        print('--template_name required')
        raise click.Abort()
    client = Client()
    response = client.get(
        '/rest/templates/packages/' + package_name + '/' + package_version + '/release/templates/' + template_name)
    utils.pretty_print(response)

def get_template_package_release_model(package_name, package_version):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/templates/packages/' + package_name + '/' + package_version + '/release/model')
    utils.pretty_print(response)

def get_template_package_workingcopy_model(package_name, package_version):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/templates/packages/' + package_name + '/' + package_version + '/workingcopy/model')
    utils.pretty_print(response)

def get_templates_packages_release(package_name, package_version):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/templates/packages/' + package_name + '/' + package_version + '/release/templates')
    utils.pretty_print(response)

def get_templates_packages_workingcopy(package_name, package_version):
    if package_name is None:
        print('--package_name required')
        raise click.Abort()
    if package_version is None:
        print('--package_version required')
        raise click.Abort()
    client = Client()
    response = client.get('/rest/templates/packages/' + package_name + '/' + package_version + '/workingcopy/templates')
    utils.pretty_print(response)

def perform_search_templates_packages(terms):
    params = {}
    if terms is None:
        print('--terms required')
        raise click.Abort()
    params['terms'] = terms
    client = Client()
    response = client.post('/rest/templates/packages/perform_search', params=params)
    utils.pretty_print(response)