Merged DR v1.2.0.0

test/MysteryTestSuite.py

@@ -0,0 +1,124 @@
+import subprocess
+import sys
+import multiprocessing
+import concurrent.futures
+import argparse
+from collections import OrderedDict
+
+cpu_threads = multiprocessing.cpu_count()
+py_version = f"{sys.version_info.major}.{sys.version_info.minor}"
+
+
+def main(args=None):
+    successes = []
+    errors = []
+    task_mapping = []
+    tests = OrderedDict()
+
+    successes.append(f"Testing {args.dr} DR with {args.count} Tests" + (f" (intensity={args.tense})" if args.dr in ['basic', 'crossed'] else ""))
+    print(successes[0])
+
+    max_attempts = args.count
+    pool = concurrent.futures.ThreadPoolExecutor(max_workers=cpu_threads)
+    dead_or_alive = 0
+    alive = 0
+
+    def test(testname: str, command: str):
+        tests[testname] = [command]
+        basecommand = f"python3.8 Mystery.py --suppress_rom --suppress_meta"
+
+        def gen_seed():
+            taskcommand = basecommand + " " + command
+            return subprocess.run(taskcommand, capture_output=True, shell=True, text=True)
+
+        for x in range(1, max_attempts + 1):
+            task = pool.submit(gen_seed)
+            task.success = False
+            task.name = testname
+            task.mode = "Mystery"
+            task.cmd = basecommand + " " + command
+            task_mapping.append(task)
+
+    for i in range(0, 100):
+        test("Mystery", "--weights mystery_testsuite.yml")
+
+    from tqdm import tqdm
+    with tqdm(concurrent.futures.as_completed(task_mapping),
+              total=len(task_mapping), unit="seed(s)",
+              desc=f"Success rate: 0.00%") as progressbar:
+        for task in progressbar:
+            dead_or_alive += 1
+            try:
+                result = task.result()
+                if result.returncode:
+                    errors.append([task.name, task.cmd, result.stderr])
+                else:
+                    alive += 1
+                    task.success = True
+            except Exception as e:
+                raise e
+
+            progressbar.set_description(f"Success rate: {(alive/dead_or_alive)*100:.2f}% - {task.name}")
+
+    def get_results(testname: str):
+        result = ""
+        for mode in ['Mystery']:
+            dead_or_alive = [task.success for task in task_mapping if task.name == testname and task.mode == mode]
+            alive = [x for x in dead_or_alive if x]
+            success = f"{testname} Rate: {(len(alive) / len(dead_or_alive)) * 100:.2f}%"
+            successes.append(success)
+            print(success)
+            result += f"{(len(alive)/len(dead_or_alive))*100:.2f}%\t"
+        return result.strip()
+
+    results = []
+    for t in tests.keys():
+        results.append(get_results(t))
+
+    for result in results:
+        print(result)
+        successes.append(result)
+
+    return successes, errors
+
+
+if __name__ == "__main__":
+    successes = []
+
+    parser = argparse.ArgumentParser(add_help=False)
+    parser.add_argument('--count', default=0, type=lambda value: max(int(value), 0))
+    parser.add_argument('--cpu_threads', default=cpu_threads, type=lambda value: max(int(value), 1))
+    parser.add_argument('--help', default=False, action='store_true')
+
+    args = parser.parse_args()
+
+    if args.help:
+        parser.print_help()
+        exit(0)
+
+    cpu_threads = args.cpu_threads
+
+    for dr in [['mystery', args.count if args.count else 1, 1]]:
+
+        for tense in range(1, dr[2] + 1):
+            args = argparse.Namespace()
+            args.dr = dr[0]
+            args.tense = tense
+            args.count = dr[1]
+            s, errors = main(args=args)
+            if successes:
+                successes += [""] * 2
+            successes += s
+            print()
+
+            if errors:
+                with open(f"{dr[0]}{(f'-{tense}' if dr[0] in ['basic', 'crossed'] else '')}-errors.txt", 'w') as stream:
+                    for error in errors:
+                        stream.write(error[0] + "\n")
+                        stream.write(error[1] + "\n")
+                        stream.write(error[2] + "\n\n")
+
+    with open("success.txt", "w") as stream:
+        stream.write(str.join("\n", successes))
+
+    input("Press enter to continue")

test/customizer/rainprevent_keydoor.yaml

@@ -0,0 +1,13 @@
+meta:
+  players: 1
+settings:
+  1:
+    door_shuffle: crossed
+    mode: standard
+    shuffle: crossed
+doors:
+  1:
+    doors:
+      Hyrule Castle East Lobby N:
+        dest: Sewers Secret Room Key Door S
+        type: Key Door

test/stats/EntranceShuffleStats.py

@@ -0,0 +1,144 @@
+import RaceRandom as random
+import logging
+import time
+
+from collections import Counter, defaultdict
+
+from source.overworld.EntranceShuffle2 import link_entrances_new
+from EntranceShuffle import link_entrances
+from BaseClasses import World
+from Regions import create_regions, create_dungeon_regions
+
+
+# tested: open + crossed (lh) Mar. 17 (made changes)
+# tested: open + simple (lh) Mar. 22
+
+def run_stats():
+    # logging.basicConfig(format='%(message)s', level=logging.DEBUG)
+    random.seed(None)
+    tests = 10000
+
+    for main_mode in ['open', 'standard', 'inverted']:
+        for shuffle_mode in ['dungeonssimple', 'dungeonsfull', 'lite', 'lean',
+                             'simple', 'restricted', 'full', 'crossed']:
+            for ls in [True, False]:
+                if ls and (main_mode == 'standard' or shuffle_mode in ['dungeonssimple', 'dungeonsfull']):
+                    continue
+
+                def runner_new(world):
+                    link_entrances_new(world, 1)
+
+                def runner_old(world):
+                    link_entrances(world, 1)
+                compare_tests(tests, shuffle_mode, main_mode, ls, runner_old, runner_new)
+
+
+def run_test_stats():
+    # logging.basicConfig(format='%(message)s', level=logging.DEBUG)
+    random.seed(None)
+    tests = 10000
+
+    for main_mode in ['open']:
+        for shuffle_mode in [#'dungeonssimple', 'dungeonsfull',
+                             'simple']:#, 'restricted', 'full', 'crossed']:
+            for ls in [True, False]:
+                if ls and (main_mode == 'standard' or shuffle_mode in ['dungeonssimple', 'dungeonsfull']):
+                    continue
+
+                def runner_new(world):
+                    link_entrances_new(world, 1)
+
+                run_tests(tests, shuffle_mode, main_mode, ls, runner_new)
+
+
+def run_old_stats():
+    # logging.basicConfig(format='%(message)s', level=logging.DEBUG)
+    random.seed(None)
+    tests = 10000
+
+    for main_mode in ['open']:
+        for shuffle_mode in [#'dungeonssimple', 'dungeonsfull',
+            'simple']:#, 'restricted', 'full', 'crossed']:
+            for ls in [True, False]:
+                if ls and (main_mode == 'standard' or shuffle_mode in ['dungeonssimple', 'dungeonsfull']):
+                    continue
+
+                def runner(world):
+                    link_entrances(world, 1)
+                run_tests(tests, shuffle_mode, main_mode, ls, runner)
+
+
+def compare_tests(tests, shuffle_mode, main_mode, links=False, runner_old=None, runner_new=None):
+    print(f'Testing {shuffle_mode} {main_mode}')
+    entrance_set = set()
+    exit_set = set()
+    ctr = defaultdict(Counter)
+    start = time.time()
+    test_loop(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links, runner_old)
+    print(f'Old test took {time.time() - start}s')
+    start = time.time()
+    test_loop(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links, runner_new, -1)
+    # test_loop(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links, runner_new)
+    print(f'New test took {time.time() - start}s')
+    dump_file(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links)
+
+
+def run_tests(tests, shuffle_mode, main_mode, links=False, runner=None):
+    print(f'Testing {shuffle_mode} {main_mode}')
+    entrance_set = set()
+    exit_set = set()
+    ctr = defaultdict(Counter)
+    test_loop(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links, runner)
+
+
+def test_loop(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links, runner, value=1):
+    for i in range(0, tests):
+        if i % 1000 == 0:
+            print(f'Test {i}')
+        seed = random.randint(0, 999999999)
+        # seed = 635441530
+        random.seed(seed)
+        world = World(1, {1: shuffle_mode}, {1: 'vanilla'}, {1: 'noglitches'}, {1: main_mode}, {}, {}, {},
+                      {}, {}, {}, {}, {}, True, {}, {}, [], {})
+        world.customizer = False
+        world.shufflelinks = {1: links}
+        create_regions(world, 1)
+        create_dungeon_regions(world, 1)
+        # print(f'Linking seed {seed}')
+        # try:
+        runner(world)
+        # except Exception as e:
+        #     print(f'Failure during seed {seed} with {e}')
+        for data in world.spoiler.entrances.values():
+            ent = data['entrance']
+            ext = data['exit']
+            # drc = data['direction']
+            entrance_set.add(ent)
+            exit_set.add(ext)
+            ctr[ent][ext] += value
+
+
+def dump_file(tests, entrance_set, exit_set, ctr, shuffle_mode, main_mode, links):
+    columns = sorted(list(exit_set))
+    rows = sorted(list(entrance_set))
+    filename = f'er_stats_{shuffle_mode}_{main_mode}_{"ls" if links else "v"}.csv'
+    with open(filename, 'w') as stat_file:
+        stat_file.write(',')
+        stat_file.write(','.join(columns))
+        stat_file.write('\n')
+        for r in rows:
+            stat_file.write(f'{r},')
+            formatted = []
+            for c in columns:
+                occurance = ctr[r][c] / tests
+                formatted.append(f'{occurance:.5%}')
+            stat_file.write(','.join(formatted))
+            stat_file.write('\n')
+
+
+if __name__ == "__main__":
+    # logging.basicConfig(format='%(message)s', level=logging.DEBUG)
+    # run_tests(1, 'restricted', 'inverted', True, lambda world: link_entrances_new(world, 1))
+    # run_test_stats()
+    # run_old_stats()
+    run_stats()