145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import subprocess
|
|
import sys
|
|
import multiprocessing
|
|
import concurrent.futures
|
|
import argparse
|
|
import sqlite3
|
|
from collections import OrderedDict
|
|
from datetime import datetime
|
|
|
|
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()
|
|
|
|
max_attempts = 1 #args and args.count if args.count is not None else 1
|
|
print(f"Testing Random OWR with {max_attempts} Tests")
|
|
|
|
pool = concurrent.futures.ThreadPoolExecutor(max_workers=cpu_threads)
|
|
dead_or_alive = 0
|
|
alive = 0
|
|
|
|
|
|
basecommand = f'py Mystery.py --suppress_rom --suppress_meta --spoiler full --outputpath L:/_Work/Zelda/ROMs/Bug/Automate/{datetime.now().strftime("%y%m%d")} --weights L:/_Work/Zelda/ROMs/Bug/Automate/_test.yml'
|
|
|
|
def gen_seed():
|
|
return subprocess.run(basecommand, 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 = "OWR Random Test"
|
|
task.mode = ""
|
|
task.cmd = basecommand
|
|
task_mapping.append(task)
|
|
|
|
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:
|
|
#with open("L:/_Work/Zelda/ROMs/Bug/Automate/_error_.txt", "a") as ff:
|
|
# ff.write("seed")
|
|
for task in progressbar:
|
|
dead_or_alive += 1
|
|
try:
|
|
result = task.result()
|
|
if result.returncode:
|
|
errors.append([datetime.now(), result.stderr])
|
|
#print(result.stderr)
|
|
#print(result.stdout)
|
|
else:
|
|
alive += 1
|
|
successes.append([datetime.now(), result.stderr])
|
|
task.success = True
|
|
except Exception as e:
|
|
with open("L:/_Work/Zelda/ROMs/Bug/Automate" + datetime.now().strftime("%y%m%d") + "/_error_.txt", "a") as ff:
|
|
exc_type, exc_obj, exc_tb = sys.exc_info()
|
|
ff.write(exc_type)
|
|
ff.write(exc_tb.tb_lineno)
|
|
raise e
|
|
|
|
progressbar.set_description(f"Success rate: {(alive/dead_or_alive)*100:.2f}% - {task.name}{task.mode}")
|
|
|
|
def get_results(testname: str):
|
|
result = ""
|
|
dead_or_alive = [task.success for task in task_mapping]
|
|
alive = [x for x in dead_or_alive if x]
|
|
success = f"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 = []
|
|
results.append(get_results(""))
|
|
|
|
#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=1, type=lambda value: max(int(value), 1))
|
|
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
|
|
|
|
args = argparse.Namespace()
|
|
successes, errors = main(args=args)
|
|
# if successes:
|
|
# successes += [""] * 2
|
|
# successes += s
|
|
# print()
|
|
|
|
|
|
if errors:
|
|
with open(f"L:/_Work/Zelda/ROMs/Bug/Automate/{datetime.now().strftime('%y%m%d')}/_error_{datetime.now().strftime('%y%m%d')}.txt", 'a') as stream:
|
|
for error in errors:
|
|
stream.write("Run at: " + error[0].strftime("%Y-%m-%d %H:%M:%S") + "\n")
|
|
stream.write(error[1] + "\n\n")
|
|
stream.write("----------------------------------------------------\n")
|
|
|
|
if successes:
|
|
with open(f"L:/_Work/Zelda/ROMs/Bug/Automate/{datetime.now().strftime('%y%m%d')}/_success_{datetime.now().strftime('%y%m%d')}.txt", 'a') as stream:
|
|
for success in successes:
|
|
stream.write("Run at: " + success[0].strftime("%Y-%m-%d %H:%M:%S") + "\n")
|
|
stream.write(success[1] + "\n\n")
|
|
stream.write("----------------------------------------------------\n")
|
|
|
|
# with open(f"L:\\_Work\\Zelda\\ROMs\\Bug\\Automate\\{datetime.now().strftime('%y%m%d')}\\_success.txt", "w") as stream:
|
|
# stream.write(str.join("\n", successes))
|
|
|
|
|
|
#conn = sqlite3.connect(f"L:/_Work/Zelda/ROMs/Bug/Automate/{datetime.now().strftime('%y%m%d')}/log.db")
|
|
# conn.execute('''CREATE TABLE SEEDGEN
|
|
# (SEED INT NOT NULL,
|
|
# MYSTERY CHAR(10),
|
|
# SETTINGSCODE CHAR(20) NOT NULL,
|
|
# VERSION CHAR(10) NOT NULL,
|
|
# TIMESTAMP INT NOT NULL,
|
|
# SUCCESS INT,
|
|
# LOG TEXT);''')
|
|
|
|
#input("Press enter to continue")
|
|
|
|
#conn.close()
|
|
|
|
exit(0)
|