#!/usr/bin/env python import argparse import ast import asyncio import random async def grab_output(p: asyncio.subprocess.Process) -> float | None: out = (await p.communicate())[0].decode() if p.returncode: print("Error:", out) return None print(out.strip()) return float(out.rsplit(':')[-1]) async def run_custom_exec(p_s: str, sample: int, search: int, insertion: int) -> float | None: p = (await asyncio.create_subprocess_exec(p_s, str(sample), str(search), str(insertion), stdout=asyncio.subprocess.PIPE)) await p.wait() return await grab_output(p) async def run_exec(p_s: str, search: int, insertion: int) -> float | None: p = await asyncio.create_subprocess_exec(p_s, str(search), str(insertion), stdout=asyncio.subprocess.PIPE) await p.wait() return await grab_output(p) def grab_limit(limit: str) -> tuple[int, int]: if '-' in limit: return tuple(map(lambda x: int(x.strip()), limit.split("-", 1))) return 10, int(limit) def get_list_average(array: list[float]) -> float: return int(sum(array) / len(array) * 1000000) / 1000000 def analysis(data: dict[int, dict[int, dict[str, list[float]]]]) -> dict[int, dict[int, dict[str, float]]]: new_dict = {} for key, value in data.items(): new_dict.update({key: {}}) for key2, value2 in value.items(): new_dict[key].update( {key2: {'normal': get_list_average(value2['normal']), 'custom': get_list_average(value2['custom'])}}) return new_dict def print_excel(data: dict[int, dict[int, dict[str, list[float]]]]) -> None: result = analysis(data) print('sample', 'search', 'normal', 'custom', sep=',') for key, value in result.items(): for key2, value2 in value.items(): print(key, key2, value2['normal'], value2['custom'], sep=',') async def main(matches: argparse.Namespace) -> None: def pow2(n: int) -> int: return 2 ** n search_limit_low, search_limit_high = grab_limit(matches.limit) limit_low, limit_high = grab_limit(matches.sample_limit) total_run = 0 fault_run = 0 result = {} for limit in range(limit_low, limit_high): for search in range(search_limit_low, search_limit_high): insertion = pow2(search + random.randint(1, 3)) search = pow2(search) sample = pow2(limit) - 1 # print('\r', insertion, search, sample, end='') if sample not in result: result.update({sample: {}}) if search not in result[sample]: result[sample].update({search: {'normal': [], 'custom': []}}) for _ in range(3): retries = 3 while (normal := await run_exec(matches.exec2, search, insertion)) is None: retries -= 1 fault_run += 1 if not retries: break total_run += 4 - retries retries = 3 while (custom := await run_custom_exec(matches.exec1, sample, search, insertion)) is None: retries -= 1 fault_run += 1 if not retries: break total_run += 4 - retries result[sample][search]['normal'].append(normal) result[sample][search]['custom'].append(custom) print(fault_run, '/', total_run) print(result) print(analysis(result)) if __name__ == '__main__': arg_ = argparse.ArgumentParser() sub_ = arg_.add_subparsers(title='sub', dest='sub') run_ = sub_.add_parser('run') sub_.add_parser('excel') run_.add_argument("exec1") run_.add_argument("exec2") run_.add_argument("sample_limit") run_.add_argument("limit") parser_ = arg_.parse_args() if parser_.sub == 'run': asyncio.run(main(parser_)) elif parser_.sub == 'excel': print_excel(ast.literal_eval(input())) else: print(analysis(ast.literal_eval(input())))