Skip to content

taps.run.main

taps/run/main.py

parse_args_to_config() 

parse_args_to_config(
    argv: Sequence[str],
) -> BenchmarkConfig

Parse sequence of string arguments into a config.

Parameters:

  • argv (Sequence[str]) –

    Sequence of string arguments.

Returns:

Source code in taps/run/main.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def parse_args_to_config(argv: Sequence[str]) -> BenchmarkConfig:
    """Parse sequence of string arguments into a config.

    Args:
        argv: Sequence of string arguments.

    Returns:
        Configuration.
    """
    parser = argparse.ArgumentParser(
        description='Task Performance Suite.',
        prog='python -m taps.run',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )

    subparsers = parser.add_subparsers(
        title='Applications',
        dest='name',
        required=True,
        help='application to execute',
    )

    apps = collections.OrderedDict(sorted(get_registered_apps().items()))
    for name, config in apps.items():
        subparser = subparsers.add_parser(
            name,
            formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        )
        RunConfig.add_argument_group(subparser, argv=argv, required=True)
        ExecutorChoicesConfig.add_argument_group(
            subparser,
            argv=argv,
            required=True,
        )
        DataTransformerChoicesConfig.add_argument_group(
            subparser,
            argv=argv,
            required=False,
        )
        FilterConfig.add_argument_group(
            subparser,
            argv=argv,
            required=True,
        )
        config.add_argument_group(subparser, argv=argv, required=True)

    args = parser.parse_args(argv)
    options = vars(args)

    app_name = options['name']
    executor_config = get_executor_config(**options)
    transformer_config = get_transformer_config(**options)
    filter_config = FilterConfig(**options)
    run_config = RunConfig(**options)
    app_config = apps[app_name](**options)

    return BenchmarkConfig(
        name=app_name,
        timestamp=datetime.now(),
        executor_name=options['executor'],
        app=app_config,
        executor=executor_config,
        transformer=transformer_config,
        filter=filter_config,
        run=run_config,
    )

run() 

run(config: BenchmarkConfig) -> None

Run an application using the configuration.

This function changes the current working directory to config.run.run_dir so that all paths are relative to the current working directory.

Source code in taps/run/main.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@_cwd_run_dir
def run(config: BenchmarkConfig) -> None:
    """Run an application using the configuration.

    This function changes the current working directory to
    `config.run.run_dir` so that all paths are relative to the current
    working directory.
    """
    start = time.perf_counter()

    cwd = pathlib.Path.cwd().resolve()

    logger.log(RUN_LOG_LEVEL, f'Starting app (name={config.name})')
    logger.log(RUN_LOG_LEVEL, config)
    logger.log(RUN_LOG_LEVEL, f'Runtime directory: {cwd}')

    config_json = config.model_dump_json(exclude={'timestamp'}, indent=4)
    with open('config.json', 'w') as f:
        f.write(config_json)

    app = config.app.create_app()
    executor = config.executor.get_executor()
    data_transformer = config.transformer.get_transformer()
    data_filter = config.filter.get_filter()
    record_logger = JSONRecordLogger(config.run.task_record_file_name)
    engine = AppEngine(
        executor,
        data_transformer=data_transformer,
        data_filter=data_filter,
        record_logger=record_logger,
    )

    with contextlib.closing(app), engine:
        app.run(engine=engine, run_dir=cwd)

    runtime = time.perf_counter() - start
    logger.log(
        RUN_LOG_LEVEL,
        f'Finished app (name={config.name}, '
        f'runtime={runtime:.2f}s, tasks={engine.tasks_executed})',
    )