feat(exp): Add new analysis script

Signed-off-by: KunoiSayami <[email protected]>

1 files changed, 167 insertions, 0 deletions

diff --git a/analysis0619.py b/analysis0619.py
new file mode 100755
index 0000000..ca1c2ff
--- /dev/null
+++ b/analysis0619.py
@@ -0,0 +1,167 @@
+#!/usr/bin/env python
+import argparse
+import ast
+import asyncio
+import random
+
+
+def reverse_decimal(data: float, count: int = 6, ratio: int = 0) -> float:
+    return int(data * (10 ** count)) / (10 ** (count - ratio))
+
+
+async def grab_output(p: asyncio.subprocess.Process) -> float | None:
+    out = (await p.communicate())[0].decode()
+    if p.returncode:
+        print("Error:", p.returncode, out.strip())
+        return None
+    print(out.strip())
+    return float(out.rsplit(':')[-1])
+
+
+async def new_run(p_s: str, *args: int) -> float | None:
+    p = await asyncio.create_subprocess_exec(p_s, *list(map(str, args)), stdout=asyncio.subprocess.PIPE)
+    await p.wait()
+    return await grab_output(p)
+
+
+async def repeat_run(p_s: str, *args: int, retries: int = 3) -> tuple[float | None, int]:
+    retries_limit = retries
+    while (ret := await new_run(p_s, *args)) is None:
+        retries -= 1
+        if not retries:
+            break
+        await asyncio.sleep(1)
+    return ret, retries_limit - retries
+
+
+async def repeat_repeat_run(p_s: str, *args: int, limit: int = 3, retries: int = 3) -> tuple[list[float], int, int]:
+    sz = []
+    total_fault = 0
+    total_run = 0
+    for _ in range(limit):
+        ret, fault = await repeat_run(p_s, *args, retries=retries)
+        total_fault += fault
+        total_run += retries + 1 - fault
+        sz.append(ret)
+        await asyncio.sleep(1)
+    return sz, total_fault, total_run
+
+
+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 | None]) -> float:
+    if None in array:
+        array.remove(None)
+    # if len(array) >= 5:
+    #     array.sort()
+    #     array = array[1:-1]
+    return reverse_decimal(sum(array) / len(array))
+
+
+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 fair_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 compare(val1: float, val2: float, ret_str: bool = False) -> str | bool:
+    if ret_str:
+        return '✅' if val1 > val2 else '❌'
+    return val1 > val2
+
+
+def get_ratio(val1: float, val2: float) -> float:
+    return reverse_decimal(val1 / val2, ratio=2)
+
+
+def print_excel(data: dict[int, dict[int, dict[str, list[float]]]]) -> None:
+    result = fair_analysis(data)
+    print('sample', 'search', 'normal', 'custom', 'ratio', 'compare', sep=',')
+    for key, value in result.items():
+        for key2, value2 in value.items():
+            normal, custom = value2['normal'], value2['custom']
+            print(key, key2, normal, custom, get_ratio(custom, normal), compare(normal, custom, True),
+                  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': []}})
+            normal, fault1, run1 = await repeat_repeat_run(matches.exec2, search, insertion, matches.total_row,
+                                                           limit=matches.test_times)
+            await asyncio.sleep(1)
+            custom, fault2, run2 = await repeat_repeat_run(matches.exec1, sample, search, insertion,
+                                                           matches.total_row, limit=matches.test_times)
+            fault_run += fault1 + fault2
+            total_run += run1 + run2
+
+            result[sample][search]['normal'] = normal
+            result[sample][search]['custom'] = custom
+
+    if matches.output is None:
+        print(fault_run, '/', total_run)
+        print(result)
+        print(analysis(result))
+    else:
+        with open(matches.output, 'w') as fout:
+            fout.write(f'{fault_run} / {total_run}\n')
+            fout.write(str(result))
+            fout.write('\n')
+            fout.write(str(analysis(result)))
+            fout.write('\n')
+
+
+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")
+    run_.add_argument('--output', default=None)
+    run_.add_argument("--total-row", default=0, type=int)
+    run_.add_argument("--test-times", default=3, type=int)
+    parser_ = arg_.parse_args()
+    if parser_.sub == 'run':
+        asyncio.run(main(parser_))
+    elif parser_.sub == 'excel':
+        print_excel(ast.literal_eval(input()))
+    else:
+        print(fair_analysis(ast.literal_eval(input())))