Home of test harnesses used in LTT Labs MarkBench
"""7-Zip test script"""
import json
import logging
import os.path
import re
import time
from subprocess import Popen
import subprocess
import requests

script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

EXECUTABLE = "7zr.exe"
DOWNLOAD_URL = f"https://www.7-zip.org/a/{EXECUTABLE}"
ABS_EXECUTABLE_PATH = os.path.join(
    os.path.dirname(os.path.realpath(__file__)), EXECUTABLE)

if os.path.isfile(ABS_EXECUTABLE_PATH) is False:
    logging.info(
        "7-Zip executable not found, downloading from %s", DOWNLOAD_URL)
    r = requests.get(DOWNLOAD_URL, allow_redirects=True, timeout=120)
    with open(ABS_EXECUTABLE_PATH, 'wb') as file:
        file.write(r.content)

command = f'{ABS_EXECUTABLE_PATH}'
command = command.rstrip()
t1 = time.time()
args = ["b"]
logging.info("Starting 7-Zip benchmark! This may take a minute or so...")
process = Popen([command, "b"], cwd=os.path.dirname(
    os.path.realpath(__file__)), stdout=subprocess.PIPE)
list_of_strings = [x.decode('utf-8').rstrip('\n')
                   for x in iter(process.stdout.readlines())]
EXIT_CODE = process.wait()

SPEED_PATTERN = '^Avr:\s*([0-9]*)\s.*\|\s*([0-9]*)\s.*$'
VERSION_PATTERN = '7-Zip \(r\) (.*)\('

VERSION = ""
SPEED_C = ""
SPEED_D = ""

# Strips the newline character
for line in list_of_strings:
    if line.isspace():
        continue
    logging.info(line.strip())
    if '7-Zip' in line:
        VERSION = re.match(VERSION_PATTERN, line).group(1)
    if 'Avr:' in line:
        SPEED_C = re.match(SPEED_PATTERN, line).group(1)
        SPEED_D = re.match(SPEED_PATTERN, line).group(2)

t2 = time.time()
logging.info("Benchmark took %s seconds", round((t2 - t1), 3))
result = {
    "score": SPEED_C + " Compression (KiB/s) | " + SPEED_D + " Decompression (KiB/s)",
    "version": VERSION.strip()
}

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as file:
    file.write(json.dumps(result))

7z/7zip.py
LTTLabsOSS/markbench-tests
"""F1 22 Test script"""
import logging
import sys
import os.path
import time
import pydirectinput as user
from f1_22_utils import get_args
from f1_22_utils import get_resolution

sys.path.insert(1, os.path.join(sys.path[0], ".."))

#pylint: disable=wrong-import-position
from harness_utils.keras_service import KerasService
from harness_utils.steam import exec_steam_run_command, get_steamapps_common_path
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.misc import remove_files
from harness_utils.process import terminate_processes
#pylint: enable=wrong-import-position

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
STEAM_GAME_ID = 1692250
VIDEO_PATH = os.path.join(get_steamapps_common_path(), "videos")

skippable = [
    os.path.join(VIDEO_PATH, "attract.bk2"),
    os.path.join(VIDEO_PATH, "cm_f1_sting.bk2"),
]


def navigate_overlay():
    """Simulate inputs to navigate ingame overlay."""
    # if steam ingame overlay is disabled it will be a an okay to press
    if kerasService.look_for_word("okay", attempts=5, interval=1):
        user.press("enter")
    # if steam ingame overlay is enabled we have to press escape and enter
    elif kerasService.look_for_word("please", attempts=5, interval=1):
        user.press("esc")
        time.sleep(0.5)
        user.press("down")
        time.sleep(0.5)
        user.press("enter")
        time.sleep(0.5)

    time.sleep(3)


def navigate_menu():
    """Simulate inputs to navigate to benchmark option."""
    # Enter options
    user.press("down")
    time.sleep(0.5)
    user.press("enter")
    time.sleep(0.5)

    # Enter settings
    user.press("enter")
    time.sleep(0.5)

    # Enter graphics settings
    user.press("right")
    time.sleep(0.5)
    user.press("enter")
    time.sleep(0.5)

    # Enter benchmark options
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("enter")

    # Run benchmark!
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("enter")


def run_benchmark():
    """Runs the actual benchmark."""
    setup_start_time = time.time()
    remove_files(skippable)
    exec_steam_run_command(STEAM_GAME_ID)

    time.sleep(20)

    # press space through the warnings
    for _ in range(5):
        user.press("space")
        time.sleep(1)

    navigate_overlay()

    result = kerasService.look_for_word("options", attempts=5, interval=1)
    if not result:
        print("Didn't land on the main menu!")
        sys.exit(1)

    navigate_menu()

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)

    test_start_time = time.time()

    # sleep 3 laps
    time.sleep(330)

    navigate_overlay()

    result = kerasService.wait_for_word("results", timeout=10)
    if not result:
        logging.info(
            "Results screen was not found! Did harness not wait long enough? Or test was too long?"
        )
        sys.exit(1)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    terminate_processes("F1")
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(
    filename=f"{LOG_DIRECTORY}/harness.log",
    format=DEFAULT_LOGGING_FORMAT,
    datefmt=DEFAULT_DATE_FORMAT,
    level=logging.DEBUG,
)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger("").addHandler(console)

args = get_args()
kerasService = KerasService(
    args.keras_host, args.keras_port, os.path.join(LOG_DIRECTORY, "screenshot.jpg")
)

try:
    start_time, end_time = run_benchmark()
    width, height = get_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time),
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)

#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes("F1")
    sys.exit(1)

F1_22/f1.py
LTTLabsOSS/markbench-tests
"""Utility functions supporting F1 22 test script."""
from argparse import ArgumentParser
import os
import re

def get_resolution() -> tuple[int]:
    """Gets resolution width and height from local xml file created by game."""
    username = os.getlogin()
    config_path = f"C:\\Users\\{username}\\Documents\\My Games\\F1 22\\hardwaresettings"
    config_filename = "hardware_settings_config.xml"
    resolution = re.compile(r"<resolution width=\"(\d+)\" height=\"(\d+)\"")
    cfg = f"{config_path}\\{config_filename}"
    height = 0
    width = 0
    with open(cfg, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            height_match = resolution.search(line)
            width_match = resolution.search(line)
            if height_match is not None:
                height = height_match.group(2)
            if width_match is not None:
                width = width_match.group(1)
    return (width, height)


def get_args() -> any:
    """Retrieve parsed command line arguments."""
    parser = ArgumentParser()
    parser.add_argument(
        "--kerasHost",
        dest="keras_host",
        help="Host for Keras OCR service",
        required=True,
    )
    parser.add_argument(
        "--kerasPort",
        dest="keras_port",
        help="Port for Keras OCR service",
        required=True,
    )
    return parser.parse_args()

F1_22/f1_22_utils.py
LTTLabsOSS/markbench-tests
"""F1 23 test script"""
import logging
from argparse import ArgumentParser
import os.path
import time
import sys
import pydirectinput as user
from f1_23_utils import get_resolution

sys.path.insert(1, os.path.join(sys.path[0], ".."))

# pylint: disable=wrong-import-position
from harness_utils.steam import exec_steam_run_command, get_app_install_location
from harness_utils.keras_service import KerasService
from harness_utils.misc import remove_files
from harness_utils.process import terminate_processes
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
# pylint: enable=wrong-import-position

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "F1_23"
STEAM_GAME_ID = 2108330
VIDEO_PATH = os.path.join(get_app_install_location(STEAM_GAME_ID), "videos")

skippable = [
    os.path.join(VIDEO_PATH, "attract.bk2"),
    os.path.join(VIDEO_PATH, "cm_f1_sting.bk2")
]


def find_settings() -> any:
    """Look for and enter settings"""
    if not kerasService.look_for_word("settings", attempts=5, interval=3):
        logging.info("Didn't find settings!")
        sys.exit(1)
    user.press("enter")
    time.sleep(1.5)


def find_graphics() -> any:
    """Look for and enter graphics settings"""
    if not kerasService.look_for_word("graphics", attempts=5, interval=3):
        logging.info("Didn't find graphics!")
        sys.exit(1)
    user.press("right")
    time.sleep(0.5)
    user.press("enter")
    time.sleep(1.5)


def find_benchmark() -> any:
    """Look for and enter benchmark options"""
    if not kerasService.look_for_word("benchmark", attempts=5, interval=3):
        logging.info("Didn't find benchmark!")
        sys.exit(1)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("enter")
    time.sleep(1.5)


def find_weather() -> any:
    """Navigate to start benchmark"""
    if not kerasService.look_for_word("weather", attempts=5, interval=3):
        logging.info("Didn't find weather!")
        sys.exit(1)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("enter")
    time.sleep(2)


def navigate_startup():
    """press space through the warnings and navigate startup menus"""
    result = kerasService.wait_for_word("product", timeout=40)
    if not result:
        logging.info("Game didn't start in time. Check settings and try again.")
        sys.exit(1)

    user.press("space")
    time.sleep(1)
    user.press("space")
    time.sleep(1)
    user.press("space")
    time.sleep(4)

    # Press enter to proceed to the main menu
    result = kerasService.wait_for_word("press", interval=2, timeout=80)
    if not result:
        logging.info("Game didn't start in time. Check settings and try again.")
        sys.exit(1)

    logging.info("Hit the title screen. Continuing")
    user.press("enter")
    time.sleep(1)

    # cancel logging into ea services
    result = kerasService.wait_for_word("login", timeout=50)
    if result:
        logging.info("Cancelling logging in.")
        user.press("enter")
        time.sleep(2)


def navigate_menu():
    """Simulate inputs to navigate to benchmark option."""
    result = kerasService.wait_for_word("theatre", interval=3, timeout=60)
    if not result:
        logging.info("Didn't land on the main menu!")
        sys.exit(1)

    logging.info("Saw the options! we are good to go!")
    time.sleep(1)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("down")
    time.sleep(0.5)
    user.press("enter")
    time.sleep(2)

    find_settings()
    find_graphics()
    find_benchmark()
    find_weather() # Run benchmark!

def run_benchmark():
    """Runs the actual benchmark."""
    remove_files(skippable)
    exec_steam_run_command(STEAM_GAME_ID)
    setup_start_time = time.time()
    time.sleep(2)
    navigate_startup()
    navigate_menu()

    result = kerasService.wait_for_word("lap", interval=3, timeout=90)
    if not result:
        logging.info("Benchmark didn't start.")
        sys.exit(1)

    logging.info("Benchmark started. Waiting for benchmark to complete.")
    test_start_time = time.time()

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)

    # sleep for 3 laps
    time.sleep(350)

    result = kerasService.wait_for_word("results", interval=3, timeout=90)
    if not result:
        logging.info("Results screen was not found!" +
            "Did harness not wait long enough? Or test was too long?")
        sys.exit(1)

    logging.info("Results screen was found! Finishing benchmark.")

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(
    filename=f"{LOG_DIRECTORY}/harness.log",
    format=DEFAULT_LOGGING_FORMAT,
    datefmt=DEFAULT_DATE_FORMAT,
    level=logging.DEBUG,
)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger("").addHandler(console)

parser = ArgumentParser()
parser.add_argument(
    "--kerasHost", dest="keras_host", help="Host for Keras OCR service", required=True
)
parser.add_argument(
    "--kerasPort", dest="keras_port", help="Port for Keras OCR service", required=True
)
args = parser.parse_args()
kerasService = KerasService(
    args.keras_host, args.keras_port, os.path.join(
        LOG_DIRECTORY, "screenshot.jpg")
)

try:
    start_time, end_time = run_benchmark()
    width, height = get_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time),
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

F1_23/f1_23.py
LTTLabsOSS/markbench-tests
"""Utility functions supporting F1 23 test script."""
import os
import re


def get_resolution() -> tuple[int]:
    """Gets resolution width and height from local xml file created by game."""
    username = os.getlogin()
    config_path = f"C:\\Users\\{username}\\Documents\\My Games\\F1 23\\hardwaresettings"
    config_filename = "hardware_settings_config.xml"
    resolution = re.compile(r"<resolution width=\"(\d+)\" height=\"(\d+)\"")
    cfg = f"{config_path}\\{config_filename}"
    height = 0
    width = 0
    with open(cfg, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            height_match = resolution.search(line)
            width_match = resolution.search(line)
            if height_match is not None:
                height = height_match.group(2)
            if width_match is not None:
                width = width_match.group(1)
    return (width, height)

F1_23/f1_23_utils.py
LTTLabsOSS/markbench-tests
'''Aida64 GPGPU test script'''
import logging
import os
import subprocess
import sys

INSTALL_DIR = "C:\\Program Files\\Aida64Business\\aida64business675"
EXECUTABLE = "aida64.exe"

script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

executable = os.path.join(INSTALL_DIR, EXECUTABLE)
report_dest = os.path.join(log_dir, "report.xml")
argstr = f"/GGBENCH {report_dest}"
result = subprocess.run([executable, "/GGBENCH", report_dest], check=False)

if result.returncode > 0:
    logging.error("Aida failed with exit code {result.returncode}")
    logging.warning(result.stdout)
    logging.warning(result.stderr)
    sys.exit(1)

aida64gpgpu/aida64gpgpu.py
LTTLabsOSS/markbench-tests
"""Atomic Heart test script"""
from argparse import ArgumentParser
import logging
import os
import time
import sys
import pydirectinput as user
from utils import read_resolution

sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
from harness_utils.steam import exec_steam_run_command, get_app_install_location
from harness_utils.misc import remove_files
from harness_utils.process import terminate_processes
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.keras_service import KerasService
#pylint: enable=wrong-import-position

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
APPDATA = os.getenv("LOCALAPPDATA")
CONFIG_LOCATION = f"{APPDATA}\\AtomicHeart\\Saved\\Config\\WindowsNoEditor"
CONFIG_FILENAME = "GameUserSettings.ini"
PROCESS_NAME = "AtomicHeart"
STEAM_GAME_ID = 668580
VIDEO_PATH = os.path.join(
    get_app_install_location(STEAM_GAME_ID), "AtomicHeart", "Content", "Movies")

skippable = [
    os.path.join(VIDEO_PATH, "Launch_4k_60FPS_PS5.mp4"),
    os.path.join(VIDEO_PATH, "Launch_4k_60FPS_XboxXS.mp4"),
    os.path.join(VIDEO_PATH, "Launch_FHD_30FPS_PS4.mp4"),
    os.path.join(VIDEO_PATH, "Launch_FHD_30FPS_XboxOne.mp4"),
    os.path.join(VIDEO_PATH, "Launch_FHD_60FPS_PC_Steam.mp4")
]

user.FAILSAFE = False


def run_benchmark():
    """Starts the benchmark"""
    remove_files(skippable)
    exec_steam_run_command(STEAM_GAME_ID)
    setup_start_time = time.time()

    time.sleep(10)

    result = kerasService.wait_for_word("press", timeout=25)
    if not result:
        logging.info("Did not see start screen")
        sys.exit(1)

    user.press("space")

    result = kerasService.look_for_word("continue", attempts=20, interval=1)
    if result:
        logging.info("Continue option available, navigating accordingly.")
        user.press("s")
        time.sleep(0.5)
        user.press("d")
        time.sleep(0.5)
        user.press("f")
        time.sleep(0.5)
        user.press("space")
    else:
        logging.info("Continue option not available, navigating accordingly.")
        user.press("d")
        time.sleep(0.5)
        user.press("f")
        time.sleep(0.5)
        user.press("space")

    time.sleep(10)

    result = kerasService.wait_for_word("continue", interval=1, timeout=80)
    if not result:
        logging.info("Did not see the option to continue. Check settings and try again.")
        sys.exit(1)

    logging.info("Continue found. Continuing Run.")
    user.press("space")
    time.sleep(5)

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)
    test_start_time = time.time()

    time.sleep(230)  # wait for No Rest For the Wicked Quest

    result = kerasService.wait_for_word("wicked", interval=1, timeout=250)
    if not result:
        logging.info(
            "Wicked was not found! Did harness not wait long enough? Or test was too long?")
        sys.exit(1)

    logging.info("Wicked found. Ending Benchmark.")
    time.sleep(5)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    # Exit
    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

parser = ArgumentParser()
parser.add_argument("--kerasHost", dest="keras_host",
                    help="Host for Keras OCR service", required=True)
parser.add_argument("--kerasPort", dest="keras_port",
                    help="Port for Keras OCR service", required=True)
args = parser.parse_args()
kerasService = KerasService(args.keras_host, args.keras_port, os.path.join(
    LOG_DIRECTORY, "screenshot.jpg"))

try:
    start_time, end_time = run_benchmark()
    height, width = read_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

atomic_heart/atomicheart.py
LTTLabsOSS/markbench-tests
"""Atomic Heart utility functions"""
import os
import re

APPDATA = os.getenv("LOCALAPPDATA")
CONFIG_LOCATION = f"{APPDATA}\\AtomicHeart\\Saved\\Config\\WindowsNoEditor"
CONFIG_FILENAME = "GameUserSettings.ini"
PROCESS_NAME = "AtomicHeart"


def read_resolution():
    """Gets resolution width and height values from local file"""
    height_pattern = re.compile(r"ResolutionSizeY=(\d+)")
    width_pattern = re.compile(r"ResolutionSizeX=(\d+)")
    cfg = f"{CONFIG_LOCATION}\\{CONFIG_FILENAME}"
    height = 0
    width = 0
    with open(cfg, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            height_match = height_pattern.search(line)
            width_match = width_pattern.search(line)
            if height_match is not None:
                height = height_match.group(1)
            if width_match is not None:
                width = width_match.group(1)
    return (height, width)

atomic_heart/utils.py
LTTLabsOSS/markbench-tests
"""Blender benchmark test script"""
from argparse import ArgumentParser
import json
import logging
import os.path
import subprocess
from zipfile import ZipFile
import requests

# omit the first arg which is the script name
parser = ArgumentParser()
parser.add_argument("-s", "--scene", dest="scene",
                    help="blender scene", metavar="scene", required=True)
parser.add_argument("-d", "--device", dest="device",
                    help="device", metavar="device", required=True)
parser.add_argument("-v", "--version", dest="version",
                    help="version", metavar="version", required=True)

args = parser.parse_args()

SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(SCRIPT_DIR, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

EXECUTABLE = "benchmark-launcher-cli.exe"
ABS_EXECUTABLE_PATH = os.path.join(SCRIPT_DIR, EXECUTABLE)


def download_and_extract_cli():
    """Downloads and extracts blender benchmark CLI"""
    download_url = "https://download.blender.org/release/BlenderBenchmark2.0/launcher/benchmark-launcher-cli-3.1.0-windows.zip"
    zip_path = os.path.join(
        SCRIPT_DIR, "benchmark-launcher-cli-3.1.0-windows.zip")
    response = requests.get(download_url, allow_redirects=True, timeout=120)
    with open(zip_path, 'wb') as zip_file:
        zip_file.write(response.content)
    with ZipFile(zip_path, 'r') as zip_object:
        zip_object.extractall(path=SCRIPT_DIR)


if os.path.isfile(ABS_EXECUTABLE_PATH) is False:
    logging.info("Downloading blender benchmark CLI")
    download_and_extract_cli()


VERSION = ""
SCENE = []
DEVICE_TYPE = ""

run_array = [ABS_EXECUTABLE_PATH, "blender", "download", args.version]
process = subprocess.run(
    run_array, capture_output=True, text=True, check=False)

if process.returncode > 0:
    logging.info(process.stdout)
    logging.info(process.stderr)
    logging.error("Test failed!")

if args.scene.lower() == "all":
    SCENE = ["monster", "classroom", "junkshop"]
else:
    SCENE = [args.scene]

if args.device.lower() == "cpu":
    DEVICE_TYPE = "CPU"
else:
    run_array = [ABS_EXECUTABLE_PATH, "devices", "-b", args.version]
    process = subprocess.run(
        run_array, capture_output=True, text=True, check=False)

    if process.returncode > 0:
        logging.info(process.stdout)
        logging.info(process.stderr)
        logging.error("Test failed!")
    else:
        logging.info(process.stdout)
        logging.info(process.stderr)
        if "OPTIX" in process.stdout or "OPTIX" in process.stderr:
            DEVICE_TYPE = "OPTIX"  # nvidia
        if "CUDA" in process.stdout or "CUDA" in process.stderr:
            DEVICE_TYPE = "CUDA"  # older non rtx nvidia
        elif "HIP" in process.stdout or "HIP" in process.stderr:
            DEVICE_TYPE = "HIP"  # amd
        elif "oneAPI" in process.stdout or "oneAPI" in process.stderr:
            DEVICE_TYPE = "oneAPI"  # intel

arg_string = ["blender", "list"]
run_array = [ABS_EXECUTABLE_PATH, "benchmark"] + SCENE + \
    ["-b", args.version, "--device-type", DEVICE_TYPE, "--json"]
logging.info("Running with arguments %s", run_array)
process = subprocess.run(
    run_array, capture_output=True, text=True, check=False)

if process.returncode > 0:
    logging.error(process.stderr)
    logging.error("Test failed!")

json_array = json.loads(process.stdout)

json_report = []
for report in json_array:
    scene_report = {
        "timestamp": report['timestamp'],
        "version": report['blender_version']['version'],
        "scene": report['scene']['label'],
        "score": round(report['stats']['samples_per_minute'], 2),
        "device": report['device_info']['compute_devices'][0]['name']
    }

    logging.info(json.dumps(scene_report, indent=2))
    scene_report['version_json'] = report['blender_version']
    scene_report['results_json'] = report['stats']
    json_report.append(scene_report)

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as file:
    file.write(json.dumps(json_report))

logging.info('Test finished!')

blenderbenchmark/blender.py
LTTLabsOSS/markbench-tests
"""Counter Strike: Global Offensive test script"""
import logging
import os
import time
import sys
from argparse import ArgumentParser
import pyautogui as gui
import pydirectinput as user
from utils import (
    get_resolution,
    benchmark_folder_exists,
    download_benchmark,
    copy_benchmark,
    STEAM_GAME_ID
)

sys.path.insert(1, os.path.join(sys.path[0], '..'))

# pylint: disable=wrong-import-position
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import exec_steam_run_command
from harness_utils.keras_service import KerasService
# pylint: enable=wrong-import-position

MAP = "de_dust2"
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "csgo.exe"


def console_command(command):
    """Types out the provided command one character at a time"""
    for char in command:
        gui.press(char)
    user.press("enter")


def run_benchmark():
    """Start the benchmark"""
    if not benchmark_folder_exists():
        download_benchmark()

    copy_benchmark()
    setup_start_time = time.time()
    exec_steam_run_command(STEAM_GAME_ID)
    time.sleep(30)  # wait for game to load into main menu

    logging.info("Waiting for game to start...")
    result = kerasService.wait_for_word("news", interval=2, timeout=120)
    if not result:
        logging.info("Game didn't start in time. Check settings and try again.")
        sys.exit(1)

    logging.info('Game started. Loading map.')
    user.press("`")
    time.sleep(1)
    console_command(f"map {MAP}")
    time.sleep(5)  # wait for map to load

    logging.info("Waiting for map to load...")
    result = kerasService.wait_for_word("terrorists", interval=2, timeout=120)
    if not result:
        logging.info("Game didn't load map. Check settings and try again.")
        sys.exit(1)

    logging.info('Map loaded. Starting benchmark.')
    time.sleep(1)
    user.press("`")
    time.sleep(0.5)
    console_command("exec benchmark")
    time.sleep(0.5)
    console_command("benchmark")

    elapsed_setup_time = round((time.time() - setup_start_time), 2)
    logging.info("Harness setup took %.2f seconds", elapsed_setup_time)
    test_start_time = time.time()

    logging.info("Waiting for benchmark to run...")
    time.sleep(43) #approximate wait time

    logging.info("Waiting for benchmark results...")
    result = kerasService.wait_for_word("finished", interval=1, timeout=60)
    if not result:
        logging.info("Game didn't finish. Check settings and try again.")
        sys.exit(1)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %.2f seconds", elapsed_test_time)

    terminate_processes(PROCESS_NAME)
    return start_time, end_time


# Entry Point
setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

parser = ArgumentParser()
parser.add_argument("--kerasHost", dest="keras_host",
                    help="Host for Keras OCR service", required=True)
parser.add_argument("--kerasPort", dest="keras_port",
                    help="Port for Keras OCR service", required=True)
args = parser.parse_args()
kerasService = KerasService(args.keras_host, args.keras_port, os.path.join(
    LOG_DIRECTORY, "screenshot.jpg"))

benchmark_folder_exists()
try:
    start_time, end_time = run_benchmark()
    height, width = get_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

csgo/csgo.py
LTTLabsOSS/markbench-tests
"""Utility functions for CS:GO test script"""
import logging
import os
from pathlib import Path
import shutil
import re
import sys
from zipfile import ZipFile

import requests

sys.path.insert(1, os.path.join(sys.path[0], '..'))

# pylint: disable=wrong-import-position
from harness_utils.steam import (
    get_registry_active_user, get_steam_folder_path, get_app_install_location)
# pylint: enable=wrong-import-position

SCRIPT_DIRECTORY = Path(__file__).resolve().parent
BENCHMARK_PATH = SCRIPT_DIRECTORY.joinpath("csgo-benchmark-master", "csgo")
ZIP_NAME = "csgo-benchmark-master.zip"
ZIP_PATH = SCRIPT_DIRECTORY.joinpath(ZIP_NAME)

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
STEAM_GAME_ID = 730
CSGO_BENCHMARK_PATH = os.path.join(SCRIPT_DIRECTORY, "csgo-benchmark-master", "csgo")
CSGO_BENCHMARK_ZIP_NAME = "csgo-benchmark-master.zip"


def get_resolution():
    """Gets the resolution from a local file"""
    height_pattern = re.compile(r"\"setting.defaultresheight\"		\"(\d+)\"")
    width_pattern = re.compile(r"\"setting.defaultres\"		\"(\d+)\"")
    steam_path = get_steam_folder_path()
    user_id = get_registry_active_user()
    config_path = Path(steam_path).joinpath(
        "userdata", user_id, STEAM_GAME_ID, "local", "cfg", "video.txt")

    height = 0
    width = 0
    with open(config_path, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            height_match = height_pattern.search(line)
            width_match = width_pattern.search(line)
            if height_match is not None:
                height = height_match.group(1)
            if width_match is not None:
                width = width_match.group(1)
    return (height, width)


def benchmark_folder_exists() -> bool:
    """Check if the CSGO Benchmark has been downloaded or not"""
    return BENCHMARK_PATH.is_dir()


def download_benchmark():
    """Downloads and extracts the CSGO Benchmark scripts"""
    download_url = "https://github.com/samisalreadytaken/csgo-benchmark/archive/master.zip"
    logging.info("Downloading and extracting benchmark to %s", SCRIPT_DIRECTORY)

    if not ZIP_PATH.exists():
        response = requests.get(download_url, allow_redirects=True, timeout=180)
        with open(ZIP_PATH, "wb") as file:
            file.write(response.content)

    with ZipFile(ZIP_PATH, "r") as zip_object:
        zip_object.extractall(path=SCRIPT_DIRECTORY)


def copy_benchmark():
    """Copies the downloaded benchmark to the CSGO directory"""
    dest_dir = Path(get_app_install_location(STEAM_GAME_ID)).joinpath("csgo")
    logging.info("Copying benchmark from %s to %s", BENCHMARK_PATH, dest_dir)
    shutil.copytree(BENCHMARK_PATH, dest_dir, dirs_exist_ok = True)

csgo/utils.py
LTTLabsOSS/markbench-tests
"""Cyberpunk 2077 test script"""
import time
import logging
import sys
import os
import pyautogui as gui
import pydirectinput as user
from cyberpunk_utils import copy_no_intro_mod, get_args, read_current_resolution


sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
from harness_utils.keras_service import KerasService
from harness_utils.output import (
    setup_log_directory, write_report_json, DEFAULT_LOGGING_FORMAT, DEFAULT_DATE_FORMAT)
from harness_utils.process import terminate_processes
from harness_utils.steam import exec_steam_game
#pylint: enable=wrong-import-position

STEAM_GAME_ID = 1091500
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "cyberpunk2077.exe"


def start_game():
    """Launch the game with no launcher or start screen"""
    return exec_steam_game(STEAM_GAME_ID, game_params=["--launcher-skip", "-skipStartScreen"])


def navigate_main_menu() -> None:
    """Simulate inputs to navigate the main menu"""
    logging.info("Navigating main menu")
    result = kerasService.look_for_word("continue", attempts=10)
    if not result:
        # an account with no save game has less menu options, so just press left and enter settings
        user.press("left")
        time.sleep(0.5)
        user.press("enter")
        time.sleep(0.5)
        user.press("3")
        time.sleep(0.5)
        user.press("3")
        time.sleep(0.5)
        user.press("3")
        time.sleep(0.5)
        user.press("b")
    else:
        user.press("left")
        time.sleep(0.5)
        user.press("down")
        time.sleep(0.5)
        user.press("enter")
        time.sleep(0.5)
        user.press("3")
        time.sleep(0.5)
        user.press("3")
        time.sleep(0.5)
        user.press("3")
        time.sleep(0.5)
        user.press("b")


def run_benchmark():
    """Start the benchmark"""
    copy_no_intro_mod()

    # Start game via Steam and enter fullscreen mode
    setup_start_time = time.time()
    start_game()

    time.sleep(10)

    result = kerasService.wait_for_word("new", interval=3, timeout=60)
    if not result:
        logging.info("Did not see settings menu option.")
        sys.exit(1)

    navigate_main_menu()

    # Start the benchmark!
    setup_end_time = time.time()
    elapsed_setup_time = round(setup_end_time - setup_start_time, 2)
    logging.info("Harness setup took %f seconds", elapsed_setup_time)

    test_start_time = time.time()

    result = kerasService.wait_for_word("fps", timeout=60, interval=1)
    if not result:
        logging.info("Benchmark didn't start.")
        sys.exit(1)

    logging.info("Benchmark started. Waiting for benchmark to complete.")
    time.sleep(70)
    result = kerasService.wait_for_word("results", timeout=240, interval=3)
    if not result:
        logging.info("Did not see results screen. Mark as DNF.")
        sys.exit(1)

    test_end_time = time.time()
    elapsed_test_time = round((test_end_time - test_start_time), 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)
    gui.screenshot(os.path.join(LOG_DIRECTORY, "results.png"))
    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

args = get_args()
kerasService = KerasService(args.keras_host, args.keras_port, os.path.join(
    LOG_DIRECTORY, "screenshot.jpg"))

try:
    start_time, endtime = run_benchmark()
    resolution = read_current_resolution()
    report = {
        "resolution": f"{resolution}",
        "start_time": round((start_time * 1000)),
        "end_time": round((endtime * 1000))
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

cyberpunk2077/cyberpunk2077.py
LTTLabsOSS/markbench-tests
"""Utility functions for Cyberpunk 2077 test script"""
from argparse import ArgumentParser
import os
import logging
from pathlib import Path
import re
import shutil

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
CYBERPUNK_INSTALL_DIR = os.path.join(
    os.environ["ProgramFiles(x86)"], "Steam\\steamapps\\common\\Cyberpunk 2077")
DEFAULT_NO_INTRO_PATH = os.path.join(CYBERPUNK_INSTALL_DIR, "archive\\pc\\mod")


def get_args() -> any:
    """Returns command line arg values"""
    parser = ArgumentParser()
    parser.add_argument("--kerasHost", dest="keras_host",
                        help="Host for Keras OCR service", required=True)
    parser.add_argument("--kerasPort", dest="keras_port",
                        help="Port for Keras OCR service", required=True)
    return parser.parse_args()


def copy_no_intro_mod() -> None:
    """Copies no intro mod file"""
    src_file = os.path.join(
        SCRIPT_DIRECTORY, "basegame_no_intro_videos.archive")
    is_valid_no_intro = os.path.isfile(src_file)

    if not is_valid_no_intro:
        raise OSError(f"Can't find no intro: {src_file}")

    # Validate/create path to directory where we will copy profile to
    dest_dir: str = DEFAULT_NO_INTRO_PATH
    try:
        Path(dest_dir).mkdir(parents=True, exist_ok=True)
    except FileExistsError as err:
        logging.error(
            "Could not create rtss profiles directory - " +
            "likely due to non-directory file existing at path."
        )
        raise err

    # Copy the profile over
    destination_file = os.path.join(dest_dir, os.path.basename(src_file))
    logging.info("Copying: %s -> %s", src_file, destination_file)
    shutil.copy(src_file, destination_file)


def read_current_resolution():
    """Get resolution from local game file"""
    app_data = os.getenv("LOCALAPPDATA")
    config_location = f"{app_data}\\CD Projekt Red\\Cyberpunk 2077"
    config_filename = "UserSettings.json"
    resolution_pattern = re.compile(r"\"value\"\: \"(\d+x\d+)\"\,")
    cfg = f"{config_location}\\{config_filename}"
    resolution = 0
    with open(cfg, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            resolution_match = resolution_pattern.search(line)
            if resolution_match is not None:
                resolution = resolution_match.group(1)
    return resolution

cyberpunk2077/cyberpunk_utils.py
LTTLabsOSS/markbench-tests
'''CinebenchR23 test script'''
import subprocess
import os
import logging
import re
import json
from argparse import ArgumentParser
import sys


parser = ArgumentParser()
parser.add_argument("-p", "--test", dest="test",
                    help="test", metavar="test", required=True)
parser.add_argument("-r", "--minduration", dest="minduration",
                    help="minduration", metavar="minduration", required=False)
args = parser.parse_args()


script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

# See https://www.maxon.net/en/cinebench-tech-info for CLI options
# g_CinebenchCpu1Test=true – runs only the Single Core test procedure
# g_CinebenchCpuXTest=true – runs only the Multi Core test procedure
# g_CinebenchAllTests=true – runs all test procedures sequentially
# g_CinebenchMinimumTestDuration=100 – sets a minimum test duration of 100 seconds

test_options = {
    "singlecore": "g_CinebenchCpu1Test=true",
    "multicore": "g_CinebenchCpuXTest=true",
    "alltests": "g_CinebenchAllTests=true"
}

if args.test not in test_options:
    logging.error("Unrecognized test %s", args.test)
    sys.exit(1)

opts = [test_options[args.test]]

# isdigit() ensures the value has no decimal as well
if args.minduration is not None and not args.minduration.isdigit():
    logging.error(
        "Minimum duration must be a whole number %s", args.minduration)
    sys.exit(1)
elif args.minduration is not None:
    opts = [*opts, f"g_CinebenchMinimumTestDuration={args.minduration}"]

logging.info(opts)
result = subprocess.run(
    ["C:\\CinebenchR23\\Cinebench.exe", *opts], capture_output=True, check=False)
if result.returncode > 0:
    logging.error("Cinebench failed to run!")
    exit(result.returncode)


with open(os.path.join(log_dir, "cinebenchlog.txt"), "wb") as f:
    f.write(result.stdout)


scorepattern = re.compile(r"^CB (\d+\.\d+) \(.+\)$")
testtitlepattern = re.compile(r"Running (\w+) CPU Render Test\.\.\.")
output = result.stdout.decode()
scorestack = []
for line in reversed(output.splitlines()):
    match = scorepattern.search(line)
    if match:
        scorestack.append(match.group(1))
        continue
    match2 = testtitlepattern.search(line)
    if match2:
        scorestack.append(match2.group(1))
        continue

if len(scorestack) % 2 > 0:
    logging.warning("Could not extract scores! Too many results found")

scores = []
i = 0
while i < len(scorestack) - 1:
    scores.append(f"{scorestack[i + 1]}:{scorestack[i]}")
    i += 2

report = {
    "test": args.test,
    "score": scores
}

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as file:
    f.write(json.dumps(report))

deprecated/cinebenchr23/cinebench.py
LTTLabsOSS/markbench-tests
"""DEPRECATED: Utility functions using Open CV"""
from enum import Enum
import logging
import time
import cv2
import pyautogui as gui
import pydirectinput as user
import numpy as np
import imutils

DEFAULT_MATCH_THRESHOLD = 0.8
DEFAULT_INTERVAL = 2  # seconds
DEFAULT_TIMEOUT = 60  # seconds

templates = {}


class ImageNotFoundTimeout(Exception):
    """Raised when image not found after timeout"""


class ImageNotFound(Exception):
    """Raised when image is not found"""


def get_template(name, aspect="16x9"):
    """Get template image of given aspect ratio"""
    return templates[name][aspect]


def gcd(a, b) -> int:
    """Recursively finds the greatest common divisor of two numbers"""
    return a if b == 0 else gcd(b, a % b)


def aspect_ratio(width, height) -> str:
    """Calculates an aspect ratio given a width and height"""
    denom = int(gcd(width, height))
    x = int(width / denom)
    y = int(height / denom)
    if x == 8 and y == 5:
        return "16x10"
    if x == 16 and y == 9:
        return "16x9"
    return f"{x}x{y}"


# This approach was largely inspired by the article
# https://pyimagesearch.com/2015/01/26/multi-scale-template-matching-using-python-opencv/
def locate_in_image(needle, haystack, threshold=DEFAULT_MATCH_THRESHOLD, debug=0):
    """Tries to find an image template within another image"""
    (template_height, template_width) = needle.shape[:2]

    if debug:
        cv2.imshow("Looking For", needle)
        cv2.waitKey(0)

    for scale in np.linspace(0.2, 1.0, 20)[::-1]:
        # resize the image according to the scale, and keep track
        # of the ratio of the resizing
        resized = imutils.resize(
            haystack, width=int(haystack.shape[1] * scale), inter=cv2.INTER_NEAREST)
        ratio = haystack.shape[1] / float(resized.shape[1])

        # if the resized image is smaller than the template, then break
        # from the loop
        if resized.shape[0] < template_height or resized.shape[1] < template_width:
            break

        (_, max_val, _, max_loc) = cv2.minMaxLoc(
            cv2.matchTemplate(resized, needle, cv2.TM_CCOEFF_NORMED))

        if debug:
            # draw a bounding box around the detected region
            cv2.rectangle(resized, (max_loc[0], max_loc[1]),
                (max_loc[0] + template_width, max_loc[1] + template_height), (0, 0, 255), 2)
            cv2.imshow("Searching", resized)
            cv2.waitKey(0)

        if max_val >= threshold:
            # compute the (x, y) coordinates of the bounding box based on the resized ratio
            (start_x, start_y) = (int(max_loc[0] * ratio), int(max_loc[1] * ratio))
            (end_x, end_y) = (
                int((max_loc[0] + template_width) * ratio),
                int((max_loc[1] + template_height) * ratio)
            )

            if debug:
                # draw a bounding box around the detected result and display the image
                cv2.rectangle(
                    haystack, (start_x, start_y), (end_x, end_y), (0, 0, 255), 2)
                cv2.imshow("Found", haystack)
                cv2.waitKey(0)

            return start_x, start_y
    raise ImageNotFound("Image not found on screen")


def locate_on_screen(template_name, threshold=DEFAULT_MATCH_THRESHOLD, debug=0):
    """Tries to find an image template on the screen"""
    screen = gui.screenshot()
    # pyautogui is using Pillow which is giving a format that must be adapted to work with opencv.
    screen = np.array(screen)
    screen = cv2.cvtColor(screen, cv2.COLOR_RGB2BGR)
    (height, width) = screen.shape[:2]
    ratio = aspect_ratio(width, height)
    needle = get_template(template_name, ratio)
    return needle, locate_in_image(needle, screen, threshold, debug)


def wait_for_image_on_screen(
        template_name, match_threshold=DEFAULT_MATCH_THRESHOLD,
        interval=DEFAULT_INTERVAL,
        timeout=DEFAULT_TIMEOUT):
    """Function that will wait for an image to appear on screen. This function will check every
     interval for a match that meets is greater than the match threshold. The function will raise
     an error if the image is not found within the timeout given. Will return the location
     of the image if found
     """
    start_time = time.time()
    current_time = start_time
    while not current_time - start_time > timeout:
        try:
            img, loc = locate_on_screen(template_name, match_threshold)
            return img, loc
        except ImageNotFound:
            pass
        time.sleep(interval)
        current_time = time.time()
    raise ImageNotFoundTimeout("Could not find image on screen within timeout")


class ClickType(Enum):
    """Enumerates different types of clicks"""
    SINGLE = 0  # uses .click()
    DOUBLE = 1  # uses .doubleclick()
    HARD = 2  # uses mouse.down() and mouse.up()
    AUTO_GUI = 3  # uses pyautogui instead of pydirectinput


def get_middle_of_rect(top_left_corner, height, width) -> tuple[int]:
    """Given x, y coordinates of top left corner of a rectangle and its
    width and height, calculate the coordinates of the center
    """
    center_x = top_left_corner[0] + (width / 2)
    center_y = top_left_corner[1] + (height / 2)
    return int(center_x), int(center_y)  # round to avoid fractional pixels


def wait_and_click(
        template_name, name, click_type: ClickType = ClickType.SINGLE, timeout=DEFAULT_TIMEOUT):
    """Wait to find an given image on screen and then click it"""
    logging.info("Waiting to find and click on %s", name)
    img, img_loc = wait_for_image_on_screen(template_name, timeout=timeout)
    click_loc = get_middle_of_rect(img_loc, img.shape[0], img.shape[1])
    logging.info("Click coordinates %s", (click_loc,))
    if click_type == ClickType.SINGLE:
        user.click(click_loc[0], click_loc[1])
    elif click_type == ClickType.DOUBLE:
        user.doubleClick(click_loc[0], click_loc[1])
    elif click_type == ClickType.HARD:
        user.moveTo(click_loc[0], click_loc[1])
        user.mouseDown()
        user.mouseUp()
    elif click_type == ClickType.AUTO_GUI:
        user.moveTo(click_loc[0], click_loc[1])
        gui.click()
    else:
        raise ValueError("Unknown click type")

deprecated/cv2_utils.py
LTTLabsOSS/markbench-tests
"""Utility functions for FarCry6 test script"""
import os
import cv2

# path relative to script
script_dir = os.path.dirname(os.path.realpath(__file__))
images_dir = os.path.join(script_dir, "images")
dir16x9 = os.path.join(images_dir, "16x9")
dir16x10 = os.path.join(images_dir, "16x10")

templates = {
    "options": {
        "16x10": cv2.imread(os.path.join(dir16x9, "options_menu_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "options_menu_1080.png"), cv2.IMREAD_UNCHANGED)
    },
    "benchmark": {
        "16x10": cv2.imread(os.path.join(dir16x9, "benchmark_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "benchmark_1080.png"), cv2.IMREAD_UNCHANGED)
    },
    "header": {
        "16x10": cv2.imread(os.path.join(dir16x9, "results_header_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "results_header_1080.png"), cv2.IMREAD_UNCHANGED)
    },
}

def get_resolution():
    """Stub: Return resolution"""
    return 0,0

deprecated/farcry6/far_cry_6_utils.py
LTTLabsOSS/markbench-tests
"""FarCry 6 test script"""
import os
from subprocess import Popen
import logging
import time
import sys
import pydirectinput as user

from far_cry_6_utils import get_resolution, templates

sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
import deprecated.cv2_utils
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
#pylint: enable=wrong-import-position

DEFAULT_INSTALLATION_PATH = (
    "C:\\Program Files (x86)\\Ubisoft\\Ubisoft Game Launcher\\games\\Far Cry 6\\bin")
EXECUTABLE = "FarCry6.exe"
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "FarCry6"


def start_game():
    """Starts the game process"""
    cmd = DEFAULT_INSTALLATION_PATH + '\\' + EXECUTABLE
    logging.info(cmd)
    return Popen(cmd)


def run_benchmark():
    """Starts the benchmark"""
    start_game()
    setup_start_time = time.time()

    time.sleep(50)

    user.press("space")
    user.press("space")

    deprecated.cv2_utils.wait_and_click(
        'options', 'options button', click_type=deprecated.cv2_utils.ClickType.HARD)
    time.sleep(2)
    deprecated.cv2_utils.wait_and_click(
        'benchmark', 'options button', click_type=deprecated.cv2_utils.ClickType.HARD)

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)

    test_start_time = time.time()

    time.sleep(75) # wait for benchmark to complete 60 + 15 grace
    deprecated.cv2_utils.wait_for_image_on_screen('header', 'results', interval=2, timeout=60)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

try:
    deprecated.cv2_utils.templates = templates
    start_time, end_time = run_benchmark()
    height, width = get_resolution()
    result = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", result)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

deprecated/farcry6/farcry6.py
LTTLabsOSS/markbench-tests
"""Hitman 3 test script"""
import os
import logging
import time
import sys
import pyautogui as gui

from hitman_3_utils import get_resolution, wait_for_image

sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import exec_steam_run_command
#pylint: enable=wrong-import-position

STEAM_GAME_ID = 1659040
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
STEAM_EXECUTABLE = "steam.exe"
PROCESS_NAMES = ['HITMAN3.exe', 'Launcher.exe']


def run_benchmark():
    """Starts the benchmark"""
    setup_start_time = time.time()
    exec_steam_run_command(STEAM_GAME_ID)

    options_image = os.path.dirname(os.path.realpath(__file__)) + "/images/hitman3_options.png"
    options_loc = wait_for_image(options_image, 0.7, 1, 15)
    print(f"Options button is here {options_loc}")
    click_me = gui.center(options_loc)
    print(f"Center of options button is here {click_me}")
    gui.click(click_me.x, click_me.y)

    time.sleep(2)
    gui.scroll(-1000)
    time.sleep(2)

    start_image = os.path.dirname(os.path.realpath(__file__)) + "/images/start_benchmark.png"
    start_loc = wait_for_image(start_image, 0.7, 1, 10)
    print(f"Start button is here {start_loc}")
    click_me = gui.center(start_loc)
    print(f"Center of start button is here {click_me}")
    gui.click(click_me.x, click_me.y)

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)
    test_start_time = time.time()

    hitman_title = os.path.dirname(os.path.realpath(__file__)) + "/images/hitman_header.png"
    time.sleep(150) # sleep during benchmark 140 + 10 seconds loading.
    result = wait_for_image(hitman_title, 0.7, 2, 60)
    if not result:
        logging.error("Benchmark failed to complete! Could not find end image")
        sys.exit(1)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)
    terminate_processes(*PROCESS_NAMES)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

try:
    start_time, end_time = run_benchmark()
    height, width = get_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(*PROCESS_NAMES)
    sys.exit(1)

deprecated/hitman3/hitman3.py
LTTLabsOSS/markbench-tests
"""Utility functions for Hitman 3 test script"""
import time
import winreg
import pyautogui as gui


def get_reg(name):
    """Get registry key value"""
    try:
        registry_key = winreg.OpenKey(
            winreg.HKEY_CURRENT_USER, r'SOFTWARE\IO Interactive\HITMAN3', 0, winreg.KEY_READ)
        value, _ = winreg.QueryValueEx(registry_key, name)
        winreg.CloseKey(registry_key)
        return value
    except WindowsError:
        return None


def get_resolution():
    """Get resolution from registry"""
    width = get_reg("ResolutionWidth")
    height = get_reg("ResolutionHeight")
    return (height, width)


def wait_for_image(img, match_threshold, interval, timeout):
    """Function that will wait for an image to appear on screen. This function will check every
     interval for a match that meets is greater than the match threshold. The function will raise
     an error if the image is not found within the timeout given. Will return the location
     of the image if found
     """
    start_time = time.time()
    current_time = start_time
    while not current_time - start_time > timeout:
        loc = gui.locateOnScreen(img, confidence=match_threshold)
        if loc:
            return loc
        time.sleep(interval)
        current_time = time.time()
    return None

deprecated/hitman3/hitman_3_utils.py
LTTLabsOSS/markbench-tests
"""Overwatch test script"""
import json
import logging
import os
import re
import sys
import time
import winreg
from argparse import ArgumentParser
from subprocess import Popen
import shutil
import psutil
import pyautogui as gui
import pydirectinput as user
from dotenv import load_dotenv
from overwatch_utils import templates

#pylint: disable=wrong-import-position
sys.path.append("..")
import deprecated.cv2_utils
#pylint: enable=wrong-import-position

load_dotenv(override=True)

HEIGHT_PATTERN = r"FullScreenHeight = \"(\d*)\""
WIDTH_PATTERN = r"FullScreenWidth = \"(\d*)\""
DEFAULT_INSTALLATION_PATH = "C:\\Program Files (x86)\\Overwatch\\_retail_"
EXECUTABLE = "Overwatch.exe"
WORKSHOP_CODE = "HYT6F" # same code Nvidia uses
CACHE_DIRECTORIES = [
    "C:\\ProgramData\\Blizzard Entertainment",
    "C:\\ProgramData\\Battle.net"
]


def get_documents_path():
    """Gets registry key of documents path."""
    shell_folder_key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders'
    try:
        root_handle = winreg.ConnectRegistry(None, winreg.HKEY_CURRENT_USER)
        shell_folders_handle = winreg.OpenKeyEx(root_handle, shell_folder_key)
        personal_path_key = winreg.QueryValueEx(shell_folders_handle, 'Personal')
        return personal_path_key[0]
    finally:
        root_handle.Close()


def get_settings_path() -> str:
    """Returns string path to game settings ini file."""
    return f"{get_documents_path()}\\Overwatch\\Settings\\Settings_v0.ini"


def read_current_resolution():
    """Reads resolution from local settings file."""
    hpattern = re.compile(HEIGHT_PATTERN)
    wpattern = re.compile(WIDTH_PATTERN)
    width = 0
    height = 0
    with open(get_settings_path(), encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            result = hpattern.match(line)
            if result is not None:
                height = result.group(1)

            result2 = wpattern.match(line)
            if result2 is not None:
                width = result2.group(1)
            if int(h) > 0 and int(w) > 0:
                break
    return (width, height)


def start_game():
    """Starts game executable."""
    cmd = DEFAULT_INSTALLATION_PATH + '\\' + EXECUTABLE
    logging.info(cmd)
    return Popen(cmd)


def send_input(string):
    """Sends series of key presses"""
    for char in string:
        gui.press(char)


def run_benchmark(username, password):
    """Runs the actual game benchmark"""
    start_game()
    time.sleep(15)

    # Login
    send_input(username)
    user.press("tab")
    send_input(password)
    user.press("tab")
    user.press("enter")

    # Get into play game menu
    time.sleep(15) # wait for login menu to authenticate
    user.press("down")
    time.sleep(1)
    user.press("space")

    # Get into custom game menu
    user.press("right")
    user.press("right")
    user.press("right")
    user.press("right")
    user.press("space")
    time.sleep(3)

    # tab over to Create Game
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("space")
    time.sleep(3)

    # get into create game settings
    user.press("right")
    user.press("right")
    user.press("up")
    user.press("left")
    user.press("space")
    time.sleep(3)

    # import custom map settings
    user.press("tab")
    user.press("right")
    user.press("space")
    send_input(WORKSHOP_CODE)
    user.press("tab")
    user.press("right")
    user.press("space")
    time.sleep(3)

    # start the benchmark
    user.press("escape")
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("tab")
    user.press("space")

    # give time for the benchmark to start
    # and get into it before waiting for the lobby screen
    time.sleep(30)

    deprecated.cv2_utils.wait_for_image_on_screen('start_button', interval=15, timeout=360)

    for process in psutil.process_iter():
        if "Overwatch" in process.name():
            process.terminate()


script_dir = os.path.dirname(os.path.realpath(__file__))

log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)

LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

parser = ArgumentParser()
parser.add_argument("-u", "--username", dest="username",
                    help="username", metavar="username", required=True)
parser.add_argument("-c", "--password", dest="password",
                    help="password", metavar="password", required=True)
args = parser.parse_args()

try:
    deprecated.cv2_utils.templates = templates
    for cache in CACHE_DIRECTORIES:
        try:
            shutil.rmtree(cache)
        except OSError as e:
            logging.warning("Couldn't delete %s: %s", cache, e)
    run_benchmark(args.username, args.password)
    w, h = read_current_resolution()
    report = {
        "resolution": f"{w}x{h}",
    }

    with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as report_file:
        report_file.write(json.dumps(report))

#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    for proc in psutil.process_iter():
        if "Overwatch" in proc.name():
            proc.terminate()
    sys.exit(1)

deprecated/overwatch/overwatch.py
LTTLabsOSS/markbench-tests
"""Utilities for Overwatch test script"""
import os
import cv2

# path relative to script
script_dir = os.path.dirname(os.path.realpath(__file__))
images_dir = os.path.join(script_dir, "images")
dir16x9 = os.path.join(images_dir, "16x9")
dir16x10 = os.path.join(images_dir, "16x10")

templates = {
    "start_button": {
        "16x10": cv2.imread(os.path.join(dir16x9, "start_room_button.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "start_room_button.png"), cv2.IMREAD_UNCHANGED)
    }
}

deprecated/overwatch/overwatch_utils.py
LTTLabsOSS/markbench-tests
"""Utility functions for Shadow of the Tomb Raider test script"""
import winreg
import os
import cv2

def get_reg(name) -> any:
    """Get registry key value"""
    reg_path = r'SOFTWARE\Eidos Montreal\Shadow of the Tomb Raider\Graphics'
    try:
        registry_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, reg_path, 0,
                                      winreg.KEY_READ)
        value, _ = winreg.QueryValueEx(registry_key, name)
        winreg.CloseKey(registry_key)
        return value
    except WindowsError:
        return None


def get_resolution() -> tuple[int]:
    """Get resolution from registry"""
    width = get_reg("FullscreenWidth")
    height = get_reg("FullscreenHeight")
    return (height, width)


# path relative to script
script_dir = os.path.dirname(os.path.realpath(__file__))
images_dir = os.path.join(script_dir, "images")
dir16x9 = os.path.join(images_dir, "16x9")
dir16x10 = os.path.join(images_dir, "16x10")

templates = {
    "load_menu_play": {
        "16x9": cv2.imread(os.path.join(images_dir, "load_menu_play.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(images_dir, "load_menu_play.png"), cv2.IMREAD_UNCHANGED)
    },
     "load_menu_play_orange": {
        "16x9": cv2.imread(os.path.join(images_dir, "play_orange.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(images_dir, "play_orange.png"), cv2.IMREAD_UNCHANGED)
    },
    "menu_options": {
        "16x9": cv2.imread(os.path.join(dir16x9, "options_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "options.png"), cv2.IMREAD_UNCHANGED)
    },
    "menu_options_save_game": {
        "16x9": cv2.imread(os.path.join(dir16x9, "options_1080_savedgame.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "options_savedgame.png"), cv2.IMREAD_UNCHANGED)
    },
    "menu_options_highlighted": {
        "16x9": cv2.imread(os.path.join(dir16x9, "options2_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "options2.png"), cv2.IMREAD_UNCHANGED)
    },
    "menu_options_save_game_highlighted": {
        "16x9": cv2.imread(os.path.join(dir16x9, "options2_1080_savegame.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "options2_savegame.png"), cv2.IMREAD_UNCHANGED)
    },
    "menu_graphics": {
        "16x9": cv2.imread(os.path.join(dir16x9, "graphics_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "graphics_button.png"), cv2.IMREAD_UNCHANGED)
    },
    "menu_graphics_tab": {
        "16x9": cv2.imread(os.path.join(dir16x9, "graphics_tab_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "graphics_tab.png"), cv2.IMREAD_UNCHANGED)
    },
    "run_benchmark": {
        "16x9": cv2.imread(os.path.join(dir16x9, "run_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "run.png"), cv2.IMREAD_UNCHANGED)
    },
    "results_header": {
        "16x9": cv2.imread(os.path.join(dir16x9, "results_1080.png"), cv2.IMREAD_UNCHANGED),
        "16x10": cv2.imread(os.path.join(dir16x10, "results.png"), cv2.IMREAD_UNCHANGED)
    }
}

deprecated/shadowofthetombraider/shadow_of_the_tomb_raider_utils.py
LTTLabsOSS/markbench-tests
"""Shadow of the Tomb Raider test script"""
import os
import logging
import time
import sys
import pydirectinput as user
import pyautogui as gui
from shadow_of_the_tomb_raider_utils import get_resolution, templates

sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
import deprecated.cv2_utils
from harness_utils.output import (
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import exec_steam_run_command
#pylint: enable=wrong-import-position

STEAM_GAME_ID = 750920
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")


def click_options(options):
    """
    If the game is freshly installed the main menu has less options thus the options button
    looks different. We also account for if the options button is highlighted if the mouse
    is hovering over it.
    """
    if options is None or len(options) == 0:
        return

    try:
        deprecated.cv2_utils.wait_and_click(
            options[0], "graphics options", deprecated.cv2_utils.ClickType.HARD)
    except deprecated.cv2_utils.ImageNotFoundTimeout:
        click_options(options[1:])


def run_benchmark():
    """Start game via Steam and enter fullscreen mode"""
    setup_start_time = time.time()
    exec_steam_run_command(STEAM_GAME_ID)

    try:
        deprecated.cv2_utils.wait_and_click('load_menu_play', "play button", timeout=30)
    except deprecated.cv2_utils.ImageNotFoundTimeout:
        deprecated.cv2_utils.wait_and_click('load_menu_play_orange', "play button", timeout=30)

    # Wait for game to load and enter graphics submenu
    option_images = [
        'menu_options_save_game',
        'menu_options_save_game_highlighted',
        'menu_options',
        'menu_options_highlighted',
    ]
    click_options(option_images)
    deprecated.cv2_utils.wait_and_click('menu_graphics', "graphics options", deprecated.cv2_utils.ClickType.HARD)
    time.sleep(2)  # let the menu transition
    gui.screenshot(os.path.join(LOG_DIRECTORY, "display_settings.png"))
    deprecated.cv2_utils.wait_and_click('menu_graphics_tab', "graphics tab", deprecated.cv2_utils.ClickType.HARD)
    gui.screenshot(os.path.join(LOG_DIRECTORY, "graphics_settings.png"))

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)
    user.press("r")
    test_start_time = time.time()

    # Wait for benchmark to complete
    time.sleep(180)
    deprecated.cv2_utils.wait_for_image_on_screen('results_header', interval=2, timeout=60)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)
    gui.screenshot(os.path.join(LOG_DIRECTORY, "results.png"))

    # Exit
    terminate_processes("SOTTR")
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

try:
    deprecated.cv2_utils.templates = templates
    start_time, end_time = run_benchmark()
    height, width = get_resolution()
    result = {
        "resolution": f"{width}x{height}",
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", result)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes("SOTTR")
    sys.exit(1)

deprecated/shadowofthetombraider/shadowofthetombraider.py
LTTLabsOSS/markbench-tests
"""Tiny Tina's Wonderland test script"""
import logging
import time
import os
from subprocess import Popen
import sys
from utils import read_resolution, try_install_paths, get_local_drives, templates

sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
import deprecated.cv2_utils
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import exec_steam_run_command
#pylint: enable=wrong-import-position

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
STEAM_GAME_ID = 1286680
EXECUTABLE = "Wonderlands.exe"


def start_game() -> any:
    """Start the game process"""
    game_process = None
    try:
        exec_path = try_install_paths(get_local_drives())
        game_process = Popen(exec_path)
    except ValueError:
        logging.info("Could not find executable, trying a steam launch")

    if game_process is None:
        game_process = exec_steam_run_command(STEAM_GAME_ID)


def run_benchmark():
    """Start the benchmark"""
    start_game()
    setup_start_time = time.time()

    # wait for menu to load
    time.sleep(30)

    deprecated.cv2_utils.wait_and_click(
        "options", "options menu", deprecated.cv2_utils.ClickType.HARD)
    deprecated.cv2_utils.wait_and_click(
        "benchmark", "benchmark menu", deprecated.cv2_utils.ClickType.HARD)
    deprecated.cv2_utils.wait_and_click(
        "start", "start benchmark", deprecated.cv2_utils.ClickType.HARD)
    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)

    test_start_time = time.time()

    time.sleep(124)
    deprecated.cv2_utils.wait_for_image_on_screen("options", 0.8, 2, 60)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)
    terminate_processes("Wonderlands")
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

try:
    deprecated.cv2_utils.templates = templates
    start_time, end_time = run_benchmark()
    height, width = read_resolution()
    result = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", result)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes("Wonderlands")
    sys.exit(1)

deprecated/tinytinaswonderland/tinytinaswonderland.py
LTTLabsOSS/markbench-tests
"""Utility functions for Tiny Tina's Wonderland test script"""
import os
import logging
import re
import winreg
import win32api
import win32file
import cv2

EXECUTABLE = "Wonderlands.exe"


def get_documents_path() -> str:
    """Use registry to find documents path"""
    shell_folder_key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders'
    try:
        root_handle = winreg.ConnectRegistry(None, winreg.HKEY_CURRENT_USER)
        shell_folders_handle = winreg.OpenKeyEx(root_handle, shell_folder_key)
        personal_path_key = winreg.QueryValueEx(shell_folders_handle, 'Personal')
        return personal_path_key[0]
    finally:
        root_handle.Close()


def valid_filepath(path: str) -> bool:
    """Validate given path is valid and leads to an existing file. A directory
    will throw an error, path must be a file"""
    if path is None or len(path.strip()) <= 0:
        return False
    if os.path.isdir(path) is True:
        return False
    return os.path.isfile(path)


def get_local_drives() -> list[str]:
    """Returns a list containing letters from local drives"""
    drive_list = win32api.GetLogicalDriveStrings()
    drive_list = drive_list.split("\x00")[0:-1]  # the last element is ""
    list_local_drives = []
    for letter in drive_list:
        if win32file.GetDriveType(letter) == win32file.DRIVE_FIXED:
            list_local_drives.append(letter)
    return list_local_drives


def try_install_paths(drives) -> str:
    """Look for executable in common install paths"""
    for drive_letter in drives:
        try_me = (
            f"{drive_letter}Program Files\\Epic Games\\TinyTinasWonderlands"
            f"\\OakGame\\Binaries\\Win64\\{EXECUTABLE}"
        )
        try_me2 = (
            f"{drive_letter}Epic Games\\TinyTinasWonderlands"
            f"\\OakGame\\Binaries\\Win64\\{EXECUTABLE}"
        )
        if valid_filepath(try_me):
            return try_me
        if valid_filepath(try_me2):
            return try_me2
    raise ValueError("Cannot find the install path for the game!")


def read_resolution() -> tuple[int]:
    """Read resolution from local ini file"""
    dest = (
        f"{get_documents_path()}\\My Games\\Tiny Tina's Wonderlands" 
        "\\Saved\\Config\\WindowsNoEditor\\GameUserSettings.ini"
    )
    hpattern = re.compile(r"ResolutionSizeY=(\d*)")
    wpattern = re.compile(r"ResolutionSizeX=(\d*)")
    h = w = 0
    with open(dest, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            result = hpattern.match(line)
            if result is not None:
                h = result.group(1)

            result2 = wpattern.match(line)
            if result2 is not None:
                w = result2.group(1)
            if int(h) > 0 and int(w) > 0:
                break
    logging.info("Current resolution is %dx%d", h, w)
    return (h, w)


# path relative to script
script_dir = os.path.dirname(os.path.realpath(__file__))
images_dir = os.path.join(script_dir, "images")
dir16x9 = os.path.join(images_dir, "16x9")
dir16x10 = os.path.join(images_dir, "16x10")

templates = {
    "options": {
        "16x10": cv2.imread(os.path.join(dir16x10, "options.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "options.png"), cv2.IMREAD_UNCHANGED)
    },
    "benchmark": {
        "16x10": cv2.imread(os.path.join(dir16x10, "benchmark.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "benchmark.png"), cv2.IMREAD_UNCHANGED)
    },
    "start": {
        "16x10": cv2.imread(os.path.join(dir16x10, "start.png"), cv2.IMREAD_UNCHANGED),
        "16x9": cv2.imread(os.path.join(dir16x9, "start.png"), cv2.IMREAD_UNCHANGED)
    }
}

deprecated/tinytinaswonderland/utils.py
LTTLabsOSS/markbench-tests
"""FLAC test script"""
import json
import logging
import os.path
import sys
import time
from subprocess import Popen

sys.path.insert(1, os.path.join(sys.path[0], ".."))

# pylint: disable=wrong-import-position
from flac_utils import download_flac, flac_folder_exists
from harness_utils.output import DEFAULT_LOGGING_FORMAT, setup_log_directory

script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
setup_log_directory(log_dir)
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

FLAC_URL = "https://ftp.osuosl.org/pub/xiph/releases/flac/flac-1.4.3-win.zip"
if flac_folder_exists() is False:
    download_flac(FLAC_URL)

EXECUTABLE = "encode-flac.bat"
ABS_EXECUTABLE_PATH = os.path.join(
    os.path.dirname(os.path.realpath(__file__)), EXECUTABLE)
command = f'{ABS_EXECUTABLE_PATH}'

command = command.rstrip()
start_time = time.time()
process = Popen(executable=command, args=[],
                cwd=os.path.dirname(os.path.realpath(__file__)))
EXIT_CODE = process.wait()

if EXIT_CODE > 0:
    logging.error("Test failed!")
    sys.exit(EXIT_CODE)

end_time = time.time()
score = round((end_time - start_time), 3)
logging.info("Benchmark took %s seconds", score)
if EXIT_CODE > 0:
    logging.error("Test failed!")
    exit(EXIT_CODE)

report = {
    "score": score,
    "version": "1.4.3"
}

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as report_file:
    report_file.write(json.dumps(report))

flac/flac.py
LTTLabsOSS/markbench-tests
"""Collection of functions to assist in running flac test script"""
import os
from zipfile import ZipFile

import requests

FLAC_FOLDER_NAME =  "flac-1.4.3-win"
FLAC_ZIP_NAME = "flac-1.4.3-win.zip"
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))

def flac_folder_exists() -> bool:
    """Check if flac has been downloaded or not"""
    return os.path.isdir(os.path.join(SCRIPT_DIR, FLAC_FOLDER_NAME))


def download_flac(url: str) -> str:
    """Download and extract FLAC, return the path to extracted files"""
    destination = os.path.join(SCRIPT_DIR, FLAC_ZIP_NAME)
    response = requests.get(url, allow_redirects=True, timeout=120)
    with open(destination, 'wb') as file:
        file.write(response.content)
    with ZipFile(destination, 'r') as zip_object:
        zip_object.extractall(path=SCRIPT_DIR)
        
flac/flac_utils.py
LTTLabsOSS/markbench-tests
"""Forza Horizon 5 test script"""
from argparse import ArgumentParser
import logging
import os
import time
import sys
import pyautogui as gui
import pydirectinput as user
from forza5_utils import read_resolution

sys.path.insert(1, os.path.join(sys.path[0], '..'))

# pylint: disable=wrong-import-position
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.rtss import  start_rtss_process, copy_rtss_profile
from harness_utils.steam import exec_steam_run_command
from harness_utils.keras_service import KerasService
# pylint: enable=wrong-import-position

STEAM_GAME_ID = 1551360
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIR, "run")
APPDATALOCAL = os.getenv("LOCALAPPDATA")
CONFIG_LOCATION = (
    f"{APPDATALOCAL}\\ForzaHorizon5\\User_SteamLocalStorageDirectory"
    "\\ConnectedStorage\\ForzaUserConfigSelections"
)
CONFIG_FILENAME = "UserConfigSelections"
PROCESSES = ["ForzaHorizon5.exe", "RTSS.exe"]


def start_rtss():
    """Sets up the RTSS process"""
    profile_path = os.path.join(SCRIPT_DIR, "ForzaHorizon5.exe.cfg")
    copy_rtss_profile(profile_path)
    return start_rtss_process()


def run_benchmark():
    """Starts the benchmark"""
    start_rtss()
    # Give RTSS time to start
    time.sleep(10)

    exec_steam_run_command(STEAM_GAME_ID)
    setup_start_time = time.time()

    # Wait for menu to load
    time.sleep(30)

    logging.info("Waiting for start prompt...")
    result = kerasService.wait_for_word("start", timeout=30)
    if not result:
        logging.info("Game didn't start.")
        sys.exit(1)

    logging.info("Accessibilty found pressing X to continue.")
    user.press("x")
    time.sleep(2)

    result = kerasService.wait_for_word("graphics", timeout=30)
    if not result:
        logging.info("Game didn't load to the settings menu.")
        sys.exit(1)

    logging.info("Graphics found, clicking and continuing.")
    gui.moveTo(result["x"], result["y"])
    time.sleep(0.2)
    gui.mouseDown()
    time.sleep(0.2)
    gui.mouseUp()
    time.sleep(1)

    result = kerasService.wait_for_word("benchmark", timeout=12)
    if not result:
        logging.info("Didn't find benchmark in settings.")
        sys.exit(1)

    gui.mouseDown(result["x"], result["y"])
    time.sleep(0.2)
    gui.mouseUp()
    time.sleep(1)
    user.press("down")
    time.sleep(0.2)
    user.press("enter")
    time.sleep(0.2)

    result = kerasService.wait_for_word("checkpoint", timeout=360)
    if not result:
        logging.info("Benchmark didn't start.")
        sys.exit(1)

    elapsed_setup_time = round((time.time() - setup_start_time), 2)
    logging.info("Harness setup took %.2f seconds", elapsed_setup_time)

    test_start_time = time.time()

    time.sleep(95) # wait for benchmark to finish 95 seconds

    result = kerasService.wait_for_word("results", timeout=25)
    if not result:
        logging.info("Results screen was not found!")
        sys.exit(1)

    test_end_time = time.time()
    elapsed_test_time = round((test_end_time - test_start_time), 2)
    logging.info("Benchmark took %.2f seconds", elapsed_test_time)

    terminate_processes(*PROCESSES)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

parser = ArgumentParser()
parser.add_argument("--kerasHost", dest="keras_host",
                    help="Host for Keras OCR service", required=True)
parser.add_argument("--kerasPort", dest="keras_port",
                    help="Port for Keras OCR service", required=True)
args = parser.parse_args()
kerasService = KerasService(args.keras_host, args.keras_port, os.path.join(
    LOG_DIRECTORY, "screenshot.jpg"))

try:
    start_time, end_time = run_benchmark()
    width, height = read_resolution(f"{CONFIG_LOCATION}\\{CONFIG_FILENAME}")
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(*PROCESSES)
    sys.exit(1)

forza5/forza5.py
LTTLabsOSS/markbench-tests
"""Utility functions for Forza Horizon 5 test script"""
import re


def read_resolution(config_path: str) -> tuple[int]:
    """Gets the resolution from local file"""
    height_pattern = re.compile(r"<ResolutionHeight value=\"(\d+)\"/>")
    width_pattern = re.compile(r"<ResolutionWidth value=\"(\d+)\"/>")
    width = 0
    height = 0
    with open(config_path, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            height_match = height_pattern.search(line)
            width_match = width_pattern.search(line)
            if height_match is not None:
                height = height_match.group(1)
            if width_match is not None:
                width = width_match.group(1)
    return (width, height)

forza5/forza5_utils.py
LTTLabsOSS/markbench-tests
"""Allows accessing Keras Service if available."""
import json
import logging
import time
import mss
import cv2
import requests
import numpy as np

logging.getLogger("requests").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)

DEFAULT_TIMEOUT = 120.0

class KerasService():
    """Sets up connection to a Keras serivce and provides methods to use it"""
    def __init__(
            self,
            ip_addr: str,
            port: int | str,
            screenshot_path: str,
            timeout: float = DEFAULT_TIMEOUT) -> None:
        self.ip_addr = ip_addr
        self.port = str(port)
        self.url = f"http://{ip_addr}:{str(port)}/process"
        self.screenshot_path = screenshot_path
        self.timeout = timeout

    def _capture_screenshot(self) -> None:
        with mss.mss() as sct:
            monitor_1 = sct.monitors[1]  # Identify the display to capture
            screen = np.array(sct.grab(monitor_1))
            screen = cv2.cvtColor(screen, cv2.COLOR_RGB2GRAY)
            cv2.imwrite(self.screenshot_path, screen)

    def _query_service(self, word: str, report_file: any) -> any:
        try:
            keras_response = requests.post(
                self.url,
                data={"word": word},
                files={"file": report_file},
                timeout=self.timeout
            )

            if not keras_response.ok:
                return None

            if "not found" in keras_response.text:
                return None

            return json.loads(keras_response.text)
        except requests.exceptions.Timeout:
            return None

    def capture_screenshot_find_word(self, word: str) -> any:
        """Take a screenshot and try to find the given word within it."""
        self._capture_screenshot()
        with open(self.screenshot_path, "rb") as report_file:
            return self._query_service(word, report_file)

    def look_for_word(self, word: str, attempts: int = 1, interval: float = 0.0) -> bool:
        """Takes a screenshot of the monitor and searches for a given word.
        Will look for the word at a given time interval until the specified number
        of attempts is reached.
        Will return early if the query result comes back with a match.
        """
        for _ in range(attempts):
            result = self.capture_screenshot_find_word(word)
            if result is not None:
                return result
            time.sleep(interval)
        return None

    def wait_for_word(self, word: str, interval: float = 0.0, timeout: float = 0.0):
        """Takes a screenshot of the monitor and searches for a given word.
        Will look for the word at a given time interval until the specified timeout
        has been exceeded.
        Will return early if the query result comes back with a match.
        """
        search_start_time = time.time()
        while time.time() - search_start_time < timeout:
            result = self.capture_screenshot_find_word(word)
            if result is not None:
                return result
            time.sleep(interval)
        return None

harness_utils/keras_service.py
LTTLabsOSS/markbench-tests
"""Misc utility functions"""
import logging
import os

def remove_files(paths: list[str]) -> None:
    """Removes files specified by provided list of file paths.
    Does nothing for a path that does not exist.
    """
    for path in paths:
        try:
            os.remove(path)
            logging.info("Removed file: %s", path)
        except FileNotFoundError:
            pass

harness_utils/misc.py
LTTLabsOSS/markbench-tests
"""Functions related to logging and formatting output from test harnesses."""
import json
import os

DEFAULT_LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
DEFAULT_DATE_FORMAT = '%m-%d %H:%M'

def setup_log_directory(log_dir: str) -> None:
    """Creates the log directory for a harness if it does not already exist"""
    if not os.path.isdir(log_dir):
        os.mkdir(log_dir)


def write_report_json(log_dir: str, report_name: str, report_json: any) -> None:
    """Writes the json output of a harness to the log directory"""
    with open(os.path.join(log_dir, report_name), "w", encoding="utf-8") as file:
        file.write(json.dumps(report_json))


def format_resolution(width: int, height: int) -> str:
    """Given width W and height H, return string in format WxH"""
    return f"{width}x{height}"


def seconds_to_milliseconds(seconds: float | int) -> int:
    """Convert seconds to milliseconds"""
    return round((seconds * 1000))

harness_utils/output.py
LTTLabsOSS/markbench-tests
"""Functions related to managing processes"""
import psutil

def terminate_processes(*process_names: str) -> None:
    """Finds given process names and terminates them"""
    for name in process_names:
        for process in psutil.process_iter():
            if name.lower() in process.name().lower():
                process.terminate()

harness_utils/process.py
LTTLabsOSS/markbench-tests
"""Utilities related to managing and using RTSS"""
import logging
import os
from subprocess import Popen
import shutil

from pathlib import Path

DEFAULT_BASE_PATH = os.path.join(os.environ["ProgramFiles(x86)"], "RivaTuner Statistics Server")
DEFAULT_EXECUTABLE_NAME = "RTSS.exe"
DEFAULT_EXECUTABLE_PATH = os.path.join(DEFAULT_BASE_PATH, DEFAULT_EXECUTABLE_NAME)
DEFAULT_PROFILES_PATH = os.path.join(DEFAULT_BASE_PATH, "Profiles")


def copy_rtss_profile(
        profile_path: str, rtss_profiles_directory: str = DEFAULT_PROFILES_PATH) -> None:
    """Copies a RTSS profile to the RTSS profile directory"""
    # Validate path of profile we want to copy over
    is_valid_profile = os.path.isfile(profile_path)

    if not is_valid_profile:
        raise FileNotFoundError(f"Can't find provided profile: {profile_path}")

    # Validate/create path to directory where we will copy profile to
    try:
        Path(rtss_profiles_directory).mkdir(parents=True, exist_ok=True)
    except FileExistsError as err:
        logging.error(
            "Could not create rtss profiles directory" 
            + " - likely due to non-directory file existing at path."
        )
        raise err

    # Copy the profile over
    destination_file = os.path.join(rtss_profiles_directory, os.path.basename(profile_path))
    logging.info("Copying: %s -> %s", profile_path, destination_file)
    shutil.copy(profile_path, destination_file)


def start_rtss_process(executable_path: str = DEFAULT_EXECUTABLE_PATH) -> any:
    """Starts the RTSS process given path the the executable"""
    if not os.path.isfile(executable_path):
        raise OSError("RTSS is not installed!")

    logging.info(executable_path)
    return Popen(executable_path)

harness_utils/rtss.py
LTTLabsOSS/markbench-tests
"""Utility functions related to using Steam for running games."""
import logging
import winreg
from subprocess import Popen
from pathlib import Path


def get_run_game_id_command(game_id: int) -> str:
    """Returns the steam run game id command with the given game ID"""
    return "steam://rungameid/" + str(game_id)


def get_steam_folder_path() -> str:
    """Gets the path to the Steam installation directory from the SteamPath registry key"""
    reg_path = r"Software\Valve\Steam"
    reg_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, reg_path, 0, winreg.KEY_READ)
    value, _  = winreg.QueryValueEx(reg_key, "SteamPath")
    return value


def get_steam_exe_path() -> str:
    """Gets the path to the Steam executable from the SteamExe registry key"""
    reg_path = r"Software\Valve\Steam"
    reg_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, reg_path, 0, winreg.KEY_READ)
    value, _  = winreg.QueryValueEx(reg_key, "SteamExe")
    return value


def get_steamapps_common_path() -> str:
    """Returns a path to the steamapps/common folder relative to the Steam installation path"""
    return str(Path(get_steam_folder_path()) / "steamapps" / "common")


def get_registry_active_user() -> int:
    """Gets the Active Process registry key to find the Steam3 ID value of the active user.
    This is needed sometimes when looking for information from local filesystems as games
    may use this value as a folder name.
    See https://developer.valvesoftware.com/wiki/SteamID for more information about ID formats.
    """
    reg_path = r"Software\Valve\Steam\ActiveProcess"
    reg_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, reg_path, 0, winreg.KEY_READ)
    value, _  = winreg.QueryValueEx(reg_key, "ActiveUser")
    return value


def get_app_install_location(app_id: int) -> str:
    """Given the Steam App ID, Gets the install directory from the Windows Registry"""
    reg_path = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\Steam App " + str(app_id)
    registry_key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, reg_path, 0, winreg.KEY_READ)
    value, _ = winreg.QueryValueEx(registry_key, "InstallLocation")
    winreg.CloseKey(registry_key)
    return value


def exec_steam_run_command(game_id: int, steam_path=None) -> Popen:
    """Runs a game using the Steam browser protocol. The `steam_path` argument can be used to
    specify a specifc path to the Steam executable instead of relying on finding the current
    installation in the Window's registry.
    
    To launch a game with provided arguments,
    see the function `exec_steam_game`.
    """
    steam_run_arg = "steam://rungameid/" + str(game_id)
    if steam_path is None:
        steam_path = get_steam_exe_path()
    logging.info("%s %s", steam_path, steam_run_arg)
    return Popen([steam_path, steam_run_arg])


def exec_steam_game(game_id: int, steam_path=None, game_params=None) -> Popen:
    """Runs a game by providing steam executable with an array of parameters.
    The `steam_path` argument can be used to specify a specifc path to the Steam executable
    instead of relying on finding the current installation in the Window's registry.
    """
    if steam_path is None:
        steam_path = get_steam_exe_path()
    if game_params is None:
        game_params = []
    command = [steam_path, "-applaunch", str(game_id)] + game_params
    logging.info(", ".join(command))
    return Popen(command)

harness_utils/steam.py
LTTLabsOSS/markbench-tests
"""Kombustor test script"""
from argparse import ArgumentParser
from subprocess import Popen
import json
import re
import os
import logging


flags = [
    "-width=",
    "-height=",
    "-benchmark",
    "-<test name>",

    # Start the artifact scanner
    "-scan"
    "-tempgraph"

    # Write GPU data (GPU temperature, FPS, etc.) to the log file every second.
    "-log_gpu_data"

    # The score file is not updated at the end of a benchmark.
    "-update_score_file_disabled"

    # By default the log file is saved in the user’s
    # temp folder (C:\Users\USER_NAME\AppData\
    # Local\Temp). This option allows to save the log
    # file in Kombustor folder
    "-logfile_in_app_folder"
]

avail_tests = [
    "vkfurrytorus",
    "glfurrytorus",
    "vkfurrymsi",
    "glfurrymsi",
    "glfurmark1700mb",
    "glfurmark3200mb",
    "glfurmark5200mb",
    "glfurmark6500mb",
    "glmsi01burn",
    "glmsi01",
    "glmsi02cpumedium",
    "glmsi02cpumedium++",
    "glmsi02gpumedium",
    "glmsi02gpumedium++",
    "glmsi02cpuhard",
    "glmsi02gpuhard",
    "glphongdonut",
    "vkphongdonut",
    "glpbrdonut",
    "vktessyspherex32",
    "vktessyspherex16",
    "gltessyspherex32",
    "gltessyspherex16",
]

INSTALL_DIR = "C:\Program Files\Geeks3D\MSI Kombustor 4 x64"
EXECUTABLE = "MSI-Kombustor-x64.exe"

parser = ArgumentParser()
parser.add_argument("-t", "--test", dest="test",
                    help="kombuster test", metavar="test", required=True)
parser.add_argument("-r", "--resolution", dest="resolution",
                    help="resolution", metavar="resolution", required=True)
parser.add_argument("-b", "--benchmark", dest="benchmark",
                    help="benchmark mode", metavar="benchmark", required=False)
args = parser.parse_args()

if args.test not in avail_tests:
    raise ValueError(f"Error, unknown test: {args.test}")

script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

match = re.search(r"^\d+,\d+$", args.resolution)
if match is None:
    raise ValueError("Resolution value must be in format height,width")
r = args.resolution.split(",")
h = r[0]
w = r[1]

cmd = f'{INSTALL_DIR}/{EXECUTABLE}'
argstr = f"-width={w} -height={h} -{args.test} -logfile_in_app_folder "
if args.benchmark == "true":
    argstr += "-benchmark"

print(cmd)
print(argstr)
process = Popen([cmd, argstr])
EXIT_CODE = process.wait()

SCORE = "N/A"
# need to find "score => 1212 points"
pattern = re.compile(r"score => (\d+)")
log_path = os.path.join(INSTALL_DIR, "_kombustor_log.txt")
with open(log_path, encoding="utf-8") as log:
    lines = log.readlines()
    for line in reversed(lines):
        match = pattern.search(line)
        if match:
            score = match.group(1)

report = {
    "resolution": f"{w}x{h}",
    "graphics_preset": "N/A",
    "test": args.test,
    "score": score
}

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as f:
    f.write(json.dumps(report))

msikombuster/msikombuster.py
LTTLabsOSS/markbench-tests
"""Recording session test script"""
import logging
import os
import socket
import sys
import time


script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)

LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

HOST = ''
PORT = 30000
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
    s.bind((HOST, PORT))
except socket.error as msg:
    logging.error("Bind failed %s", msg)
    sys.exit(1)

s.listen(1)
while True:
    time.sleep(0.1)
    conn, addr = s.accept()
    if conn:
        sys.exit(0)
recording_session/dummy.py
LTTLabsOSS/markbench-tests
"""Utility functions supporting RDR2 test script."""
import getpass
import pathlib
import re


def get_resolution() -> tuple[int]:
    """Gets resolution width and height from local xml file created by game."""
    # C:\Users\User\Documents\Rockstar Games\Red Dead Redemption 2\Settings\system.xml"
    path = pathlib.Path(
        "C:/Users/", getpass.getuser(), 
        "Documents", "Rockstar Games", "Red Dead Redemption 2", "Settings", 
        "system.xml"
    )
    width = "0"
    height = "0"

    screen_width = re.compile(r"<screenWidth value=\"(\d+)\" />")
    screen_height = re.compile(r"<screenHeight value=\"(\d+)\" />")

    try:
        with open(path, encoding="utf-8") as file:
            lines = file.readlines()
            for line in lines:
                width_match = screen_width.search(line)
                height_match = screen_height.search(line)
                if width_match is not None:
                    width = width_match.group(1)
                if height_match is not None:
                    height = height_match.group(1)
    except OSError:
        width = "0"
        height = "0"

    return (width, height)

reddeadredemption2/red_dead_redemption_2_utils.py
LTTLabsOSS/markbench-tests
"""Red Dead Redemption 2 test script"""
import logging
import os
import time
import sys
import pydirectinput as user

from red_dead_redemption_2_utils import get_resolution

sys.path.insert(1, os.path.join(sys.path[0], '..'))

#pylint: disable=wrong-import-position
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import exec_steam_run_command
#pylint: enable=wrong-import-position

STEAM_GAME_ID = 1174180
PROCESS_NAME = "RDR2"
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")


def run_benchmark():
    """Starts the benchmark"""
    # Wait for game to load to main menu
    setup_start_time = time.time()
    exec_steam_run_command(STEAM_GAME_ID)
    time.sleep(65)

    # Press Z to enter settings
    user.press("z")
    time.sleep(3)

    # Enter graphics menu
    ## ensure we are starting from the top left of the screen
    user.press("up")
    user.press("up")
    user.press("left")
    user.press("left")

    user.press("down")
    user.press("enter")
    time.sleep(3)

    # Run benchmark by holding X for 2 seconds
    user.keyDown('x')
    time.sleep(1.5)
    user.keyUp("x")

    # Press enter to confirm benchmark
    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)
    user.press("enter")
    test_start_time = time.time()

    # Wait for the benchmark to complete
    time.sleep(315)  # 5 min, 15 seconds.
    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    # Exit
    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

try:
    start_time, end_time = run_benchmark()
    width, height = get_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

reddeadredemption2/reddeadredemption2.py
LTTLabsOSS/markbench-tests
"""Returnal test script"""
import os
import logging
import sys
import time
import pydirectinput as user

from returnal_utils import get_resolution, get_args

sys.path.insert(1, os.path.join(sys.path[0], '..'))

# pylint: disable=wrong-import-position
from harness_utils.keras_service import KerasService
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.misc import remove_files
from harness_utils.process import terminate_processes
from harness_utils.steam import (
  exec_steam_run_command,
  get_steamapps_common_path,
)
# pylint: enable=wrong-import-position

STEAM_GAME_ID = 1649240
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "Returnal"
LOCAL_USER_SETTINGS = os.path.join(
  os.getenv('LOCALAPPDATA'), "Returnal", "Steam",
  "Saved", "Config", "WindowsNoEditor", "GameUserSettings.ini"
  )
VIDEO_PATH = os.path.join(get_steamapps_common_path(), "Returnal", "Returnal", "Content", "Movies")

user.FAILSAFE = False

skippable_videos = [
    os.path.join(VIDEO_PATH, "Logos_PC.mp4"),
    os.path.join(VIDEO_PATH, "Logos_PC_UW21.mp4"),
    os.path.join(VIDEO_PATH, "Logos_PC_UW32.mp4"),
    os.path.join(VIDEO_PATH, "Logos_Short_PC.mp4"),
    os.path.join(VIDEO_PATH, "Logos_Short_PC_UW21.mp4"),
    os.path.join(VIDEO_PATH, "Logos_Short_PC_UW32.mp4"),
]

def check_vram_alert(attempts: int) -> bool:
    """Look for VRAM alert in menu"""
    logging.info("Checking for VRAM alert prompt")
    for _ in range(attempts):
        alert_result = kerasService.capture_screenshot_find_word("alert")
        locate_result = kerasService.capture_screenshot_find_word("locate")
        if locate_result is not None:
            return False
        if alert_result is not None:
            return True
        time.sleep(2)

    return False


def escape_vram_alert():
    """Navigate VRAM alert"""
    user.keyDown("space")
    time.sleep(4)
    user.keyUp("space")


def navigate_options_menu() -> None:
    """Simulate inputs to navigate to options menu"""
    logging.info("Navigating to options menu")
    user.press("esc")
    time.sleep(0.2)
    user.press("enter")
    time.sleep(0.2)
    user.press("q")
    time.sleep(0.2)
    user.keyDown("tab")
    time.sleep(5)
    user.keyUp("tab")


def run_benchmark() -> tuple[float]:
    """Run the benchmark"""
    logging.info("Removing intro videos")
    remove_files(skippable_videos)

    logging.info("Starting game")
    exec_steam_run_command(STEAM_GAME_ID)
    setup_start_time = time.time()

    time.sleep(10)

    # Check if GPU has too little v ram and skip alert
    alert_found = check_vram_alert(7)

    if alert_found:
        escape_vram_alert()

    # Make sure the game started correctly
    result = kerasService.look_for_word("locate", 10, 5)
    if not result:
        logging.info("Could not find prompt to open menu!")
        sys.exit(1)

    # Navigate to in-game benchmark and start it
    navigate_options_menu()

    setup_end_time = time.time()
    elapsed_setup_time = round((setup_end_time - setup_start_time), 2)
    logging.info("Setup took %s seconds", elapsed_setup_time)

    test_start_time = time.time()

    # Wait for benchmark to complete
    time.sleep(120)

    # Wait for results screen to display info
    result = kerasService.look_for_word("benchmark", 12, 5)
    if not result:
        logging.info(
            "Results screen was not found! Did harness not wait long enough? Or test was too long?")
        sys.exit(1)

    test_end_time = time.time()
    elapsed_test_time = round((test_end_time - test_start_time), 2)
    logging.info("Benchmark took %s seconds", elapsed_test_time)

    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

args = get_args()
kerasService = KerasService(
    args.keras_host, args.keras_port, os.path.join(LOG_DIRECTORY, "screenshot.jpg"))

try:
    start_time, end_time = run_benchmark()
    height, width = get_resolution(LOCAL_USER_SETTINGS)
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)

#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

returnal/returnal.py
LTTLabsOSS/markbench-tests
"""Utility functions supporting Returnal test script."""
from argparse import ArgumentParser
import os
import re


def get_resolution(config_path: str) -> tuple[int]:
    """Retrieve the resolution from local configuration files."""
    width_pattern = re.compile(r"ResolutionSizeX=(\d+)")
    height_pattern = re.compile(r"ResolutionSizeY=(\d+)")
    width = 0
    height = 0

    with open(config_path, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            width_match = width_pattern.match(line)
            height_match = height_pattern.match(line)

            if width_match:
                width = width_match.group(1)
            if height_match:
                height = height_match.group(1)

    return (height, width)


def remove_intro_videos(file_paths: list[str]) -> None:
    """Removes given video file paths"""
    for video in file_paths:
        try:
            os.remove(video)
        except FileNotFoundError:
            # If file not found, it has likely already been deleted before.
            pass


def get_args() -> any:
    """Get command line arguments"""
    parser = ArgumentParser()
    parser.add_argument(
        "--kerasHost", dest="keras_host", help="Host for Keras OCR service", required=True)
    parser.add_argument(
        "--kerasPort", dest="keras_port", help="Port for Keras OCR service", required=True)
    return parser.parse_args()

returnal/returnal_utils.py
LTTLabsOSS/markbench-tests
"""Superposition test script"""
from argparse import ArgumentParser
from subprocess import Popen
import json
import re
import os
import logging
import sys

avail_presets = [
    "low",
    "medium",
    "high",
    "extreme",
    "4k_optimized",
    "8k_optimized"
]

INSTALL_DIR = "C:\\Program Files\\Unigine\\Superposition Benchmark\\bin"
EXECUTABLE = "superposition_cli.exe"

parser = ArgumentParser()
parser.add_argument("-a", "--api", dest="api",
                    help="graphics api", metavar="api", required=True)
parser.add_argument("-p", "--preset", dest="preset",
                    help="performance preset", metavar="preset", required=True)
parser.add_argument("-r", "--resolution", dest="resolution",
                    help="resolution", metavar="resolution", required=True)
args = parser.parse_args()

if args.preset not in avail_presets:
    raise ValueError(f"Error, unknown preset: {args.preset}")

match = re.search(r"^\d+,\d+$", args.resolution)
if match is None:
    raise ValueError("Resolution value must be in format height,width")
r = args.resolution.split(",")
h = r[0]
w = r[1]

script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

cmd = f'{INSTALL_DIR}\\{EXECUTABLE}'
argstr = f"-fullscreen 1 -mode default -api {args.api} -quality {args.preset} -resolution {w}x{h}"
argstr += f" -log_txt {log_dir}\\log.txt"

logging.info(cmd)
logging.info(argstr)
argies = argstr.split(" ")
cmd = cmd.rstrip()
process = Popen([cmd, *argies])
EXIT_CODE = process.wait()

if EXIT_CODE > 0:
    logging.error("Test failed!")
    sys.exit(EXIT_CODE)

pattern = re.compile(r"Score: (\d+)")
log_path = os.path.join(log_dir, "log.txt")
with open(log_path, encoding="utf-8") as log:
    lines = log.readlines()
    for line in lines:
        match = pattern.search(line)
        if match:
            score = match.group(1)

report = {
    "resolution": f"{w}x{h}",
    "graphics_preset": args.preset,
    "score": score
}

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as file:
    file.write(json.dumps(report))

superposition/superposition.py
LTTLabsOSS/markbench-tests
"""The Last of Us Part I test script"""
import logging
import os
import time
import sys
import pydirectinput as user

from the_last_of_us_part_i_utils import get_args, get_resolution

sys.path.insert(1, os.path.join(sys.path[0], '..'))
#pylint: disable=wrong-import-position
from harness_utils.keras_service import KerasService
from harness_utils.output import (
    format_resolution,
    seconds_to_milliseconds,
    setup_log_directory,
    write_report_json,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import (
  get_registry_active_user,
  exec_steam_run_command,
)
#pylint: enable=wrong-import-position

STEAM_GAME_ID = 1888930
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "tlou"

user.FAILSAFE = False


def navigate_main_menu() -> None:
    """Input to navigate main menu"""
    logging.info("Navigating main menu")

    # Enter TLOU menu
    user.press("space")
    time.sleep(0.5)
    user.press("space")
    time.sleep(0.5)
    # Press load game
    user.press("s")
    time.sleep(0.5)
    user.press("s")
    time.sleep(0.5)
    user.keyDown("space")
    time.sleep(0.5)
    # Go to bottom save
    user.press("w")
    time.sleep(0.5)
    user.press("space")
    time.sleep(0.5)
    user.press("space")
    time.sleep(2)


def run_benchmark():
    """Starts the benchmark"""
    exec_steam_run_command(STEAM_GAME_ID)
    setup_start_time = time.time()

    time.sleep(10)

    result = kerasService.wait_for_word("press", interval=3, timeout=60)
    if not result:
        logging.info("Did not see start screen")
        sys.exit(1)

    navigate_main_menu()

    # press load save
    result = kerasService.look_for_word("yes", attempts=10, interval=1)
    if not result:
        logging.info("Did not load the save")
        sys.exit(1)

    user.press("a")
    time.sleep(0.5)
    user.press("space")

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)
    test_start_time = time.time()

    # wait for black screen
    time.sleep(150)

    # This actually looks for "from?" but the current ML model sees it as fromy
    result = kerasService.wait_for_word("fromy", timeout=250)
    if not result:
        logging.info("Did not find prompt to end harness.")
        sys.exit(1)

    # Wait for black screen
    time.sleep(24)

    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

args = get_args()
kerasService = KerasService(
    args.keras_host, args.keras_port, os.path.join(LOG_DIRECTORY, "screenshot.jpg"))

try:
    start_time, end_time = run_benchmark()
    steam_id = get_registry_active_user()
    config_path = os.path.join(
        os.environ["HOMEPATH"], "Saved Games" ,"The Last of Us Part I",
        "users", str(steam_id), "screeninfo.cfg"
    )
    height, width = get_resolution(config_path)
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(end_time)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

the_last_of_us_part_i/the_last_of_us_part_i.py
LTTLabsOSS/markbench-tests
"""Utility functions for The Last of Us Part I test script"""
from argparse import ArgumentParser
import ctypes
import re


def get_args() -> any:
    """Get command line arg values"""
    parser = ArgumentParser()
    parser.add_argument(
        "--kerasHost", dest="keras_host", help="Host for Keras OCR service", required=True)
    parser.add_argument(
        "--kerasPort", dest="keras_port", help="Port for Keras OCR service", required=True)
    return parser.parse_args()


def get_windowed_resolution(lines: list[str]):
    """Get windowed resolution value from local game file"""
    width_pattern = re.compile(r"WindowWidth=(\d+)")
    height_pattern = re.compile(r"WindowHeight=(\d+)")

    window_width = 0
    window_height = 0

    for line in lines:
        width_match = width_pattern.search(line)
        height_match = height_pattern.search(line)

        if width_match:
            window_width = width_match.group(1)

        if height_match:
            window_height = height_match.group(1)

    return window_height, window_width


def get_borderless_resolution(lines: list[str]):
    """Get borderless resolution value from local game file

    Note:
    If resolution is set to native resolution, then the values of
    BorderlessWidth and BorderlessHeight saved in the config are 0, 0.
    Without getting into various 3rd-party gui libraries, we can grab
    resolution sizes from the system metrics if we get values of 0.
    However, this has limitations: This can only get values for the primary monitor
    and reportedly may not return correct values with high DPI displays.
    """

    width_pattern = re.compile(r"BorderlessWidth=(\d+)")
    height_pattern = re.compile(r"BorderlessHeight=(\d+)")

    window_width = 0
    window_height = 0

    user32 = ctypes.windll.user32

    for line in lines:
        width_match = width_pattern.search(line)
        height_match = height_pattern.search(line)

        if width_match:
            window_width = width_match.group(1)
            if window_width == 0:
                window_width = user32.GetSystemMetrics(0)

        if height_match:
            window_height = height_match.group(1)
            if window_height == 0:
                window_height = user32.GetSystemMetrics(1)

    return window_height, window_width


def get_resolution(config_path: str):
    """Gets the resolution by retrieving the active steam user id from
    registry to complete the path to the local config file.
    Then parses the local config file and grabbing values depending on
    if the display settings were set to Windowed or Borderless windowed mode.
    """
    window_mode_pattern = re.compile(r"WindowMode=(\d)")

    window_mode = None
    window_width = 0
    window_height = 0

    with open(config_path, encoding="utf-8") as file:
        window_mode_line = file.readline()
        window_mode = window_mode_pattern.search(window_mode_line).group(1)

        if not window_mode:
            return window_height, window_width

        lines = file.readlines()
        if int(window_mode) == 1:
            window_height, window_width = get_borderless_resolution(lines)
        else:
            window_height, window_width = get_windowed_resolution(lines)

    return window_height, window_width

the_last_of_us_part_i/the_last_of_us_part_i_utils.py
LTTLabsOSS/markbench-tests
"""Total War: Warhammer III test script"""
from argparse import ArgumentParser
import logging
import os
import time
import sys
import pyautogui as gui
import pydirectinput as user
from twwh3_utils import read_current_resolution

sys.path.insert(1, os.path.join(sys.path[0], '..'))
# pylint: disable=wrong-import-position
from harness_utils.process import terminate_processes
from harness_utils.output import (
    format_resolution,
    setup_log_directory,
    write_report_json,
    seconds_to_milliseconds,
    DEFAULT_LOGGING_FORMAT,
    DEFAULT_DATE_FORMAT
)
from harness_utils.steam import get_app_install_location
from harness_utils.keras_service import KerasService
# pylint: enable=wrong-import-position

SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
PROCESS_NAME = "Warhammer3.exe"
STEAM_GAME_ID = 1142710

user.FAILSAFE = False


def start_game():
    """Starts the game process"""
    cmd_string = f"start /D \"{get_app_install_location(STEAM_GAME_ID)}\" {PROCESS_NAME}"
    logging.info(cmd_string)
    return os.system(cmd_string)


def skip_logo_screens() -> None:
    """Simulate input to skip logo screens"""
    logging.info("Skipping logo screens")

    # Enter TWWH3 menu
    user.press("escape")
    time.sleep(0.5)
    user.press("escape")
    time.sleep(0.5)
    user.press("escape")
    time.sleep(0.5)
    user.press("escape")
    time.sleep(0.5)
    user.press("escape")
    time.sleep(0.5)
    user.press("escape")
    time.sleep(0.5)
    user.press("escape")
    time.sleep(0.5)


def run_benchmark():
    """Starts the benchmark"""
    start_game()
    setup_start_time = time.time()
    time.sleep(5)

    result = kerasService.look_for_word("warning", attempts=10, interval=5)
    if not result:
        logging.info("Did not see warnings. Did the game start?")
        sys.exit(1)

    skip_logo_screens()
    time.sleep(2)

    result = kerasService.look_for_word("options", attempts=10, interval=1)
    if not result:
        logging.info("Did not find the options menu. Did the game skip the intros?")
        sys.exit(1)

    gui.moveTo(result["x"], result["y"])
    time.sleep(0.2)
    gui.mouseDown()
    time.sleep(0.2)
    gui.mouseUp()

    result = kerasService.look_for_word("ad", attempts=10, interval=1)
    if not result:
        logging.info("Did not find the advanced menu. Did the game skip the intros?")
        sys.exit(1)

    gui.moveTo(result["x"], result["y"])
    time.sleep(0.2)
    gui.mouseDown()
    time.sleep(0.2)
    gui.mouseUp()

    result = kerasService.look_for_word("bench", attempts=10, interval=1)
    if not result:
        logging.info("Did not find the benchmark menu. Did the game skip the intros?")
        sys.exit(1)

    gui.moveTo(result["x"], result["y"])
    time.sleep(0.2)
    gui.mouseDown()
    time.sleep(0.2)
    gui.mouseUp()
    time.sleep(2)
    user.press("enter")

    result = kerasService.wait_for_word("fps", interval=5, timeout=100)
    if not result:
        logging.info("Benchmark didn't start.")
        sys.exit(1)

    elapsed_setup_time = round(time.time() - setup_start_time, 2)
    logging.info("Setup took %f seconds", elapsed_setup_time)
    test_start_time = time.time()

    time.sleep(100)  # Wait for benchmark

    result = kerasService.wait_for_word("fps", interval=5, timeout=250)
    if not result:
        logging.info(
            "Results screen was not found! Did harness not wait long enough? Or test was too long?")
        sys.exit(1)

    # Wait 5 seconds for benchmark info
    time.sleep(5)

    # End the run
    test_end_time = time.time()
    elapsed_test_time = round(test_end_time - test_start_time, 2)
    logging.info("Benchmark took %f seconds", elapsed_test_time)

    # Exit
    terminate_processes(PROCESS_NAME)
    return test_start_time, test_end_time


setup_log_directory(LOG_DIRECTORY)

logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
                    format=DEFAULT_LOGGING_FORMAT,
                    datefmt=DEFAULT_DATE_FORMAT,
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

parser = ArgumentParser()
parser.add_argument("--kerasHost", dest="keras_host",
                    help="Host for Keras OCR service", required=True)
parser.add_argument("--kerasPort", dest="keras_port",
                    help="Port for Keras OCR service", required=True)
args = parser.parse_args()
kerasService = KerasService(args.keras_host, args.keras_port, os.path.join(
    LOG_DIRECTORY, "screenshot.jpg"))

try:
    start_time, endtime = run_benchmark()
    height, width = read_current_resolution()
    report = {
        "resolution": format_resolution(width, height),
        "start_time": seconds_to_milliseconds(start_time),
        "end_time": seconds_to_milliseconds(endtime)
    }

    write_report_json(LOG_DIRECTORY, "report.json", report)
#pylint: disable=broad-exception-caught
except Exception as e:
    logging.error("Something went wrong running the benchmark!")
    logging.exception(e)
    terminate_processes(PROCESS_NAME)
    sys.exit(1)

total_war_warhammer_iii/twwh3.py
LTTLabsOSS/markbench-tests
"""Utility functions for Total War: Warhammer III test script"""
import os
import re

APPDATA = os.getenv("APPDATA")
CONFIG_LOCATION = f"{APPDATA}\\The Creative Assembly\\Warhammer3\\scripts"
CONFIG_FILENAME = "preferences.script.txt"

def read_current_resolution():
    """Reads resolutions settings from local game file"""
    height_pattern = re.compile(r"y_res (\d+);")
    width_pattern = re.compile(r"x_res (\d+);")
    cfg = f"{CONFIG_LOCATION}\\{CONFIG_FILENAME}"
    height_value = 0
    width_value = 0
    with open(cfg, encoding="utf-8") as file:
        lines = file.readlines()
        for line in lines:
            height_match = height_pattern.search(line)
            width_match = width_pattern.search(line)
            if height_match is not None:
                height_value = height_match.group(1)
            if width_match is not None:
                width_value = width_match.group(1)
    return (height_value, width_value)
total_war_warhammer_iii/twwh3_utils.py
LTTLabsOSS/markbench-tests
"""Test script for y-cruncher"""
import json
import logging
import os.path
import sys
import re
import glob
from subprocess import Popen

sys.path.insert(1, os.path.join(sys.path[0], ".."))

# pylint: disable=wrong-import-position
from ycruncher_utils import YCRUNCHER_FOLDER_NAME, download_ycruncher, ycruncher_folder_exists


ABS_EXECUTABLE_PATH = os.path.join(os.path.dirname(
    os.path.realpath(__file__)), YCRUNCHER_FOLDER_NAME, "y-cruncher.exe")

script_dir = os.path.dirname(os.path.realpath(__file__))
log_dir = os.path.join(script_dir, "run")
if not os.path.isdir(log_dir):
    os.mkdir(log_dir)
LOGGING_FORMAT = '%(asctime)s %(levelname)-s %(message)s'
logging.basicConfig(filename=f'{log_dir}/harness.log',
                    format=LOGGING_FORMAT,
                    datefmt='%m-%d %H:%M',
                    level=logging.DEBUG)
console = logging.StreamHandler()
formatter = logging.Formatter(LOGGING_FORMAT)
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

if ycruncher_folder_exists() is False:
    download_ycruncher()

# omit the first arg which is the script name
args = sys.argv[1:]
logging.info(args)
command = f'{ABS_EXECUTABLE_PATH}'
command = command.rstrip()
arg_string = ['skip-warnings', 'bench', '5b', '-o',
              os.path.join(os.path.dirname(os.path.realpath(__file__)), 'run')]

logging.info(arg_string)
process = Popen(executable=command, args=arg_string)
EXIT_CODE = process.wait()

if EXIT_CODE > 0:
    logging.error("Test failed!")
    sys.exit(EXIT_CODE)

list_of_files = glob.glob(os.path.join(
    os.path.dirname(os.path.realpath(__file__)), 'run', '*.txt'))
latest_file = max(list_of_files, key=os.path.getctime)
print(latest_file)

with open(os.path.join(os.path.dirname(
    os.path.realpath(__file__)), 'run', latest_file), "r", encoding="utf-8") as file:
    Lines = file.readlines()

TIME_PATTERN = r'^.*:\s*(.*) seconds$'
TUNE_PATTERN = r'^.*:\s*(.*)$'
TIME = ""
TUNING = ""

# Strips the newline character
for line in Lines:
    if 'Total Computation Time' in line:
        time = re.match(TIME_PATTERN, line).group(1)
    if 'Tuning:' in line:
        tuning = re.match(TUNE_PATTERN, line).group(1)

report = {
    "score": time,
    "test": tuning
}

with open(os.path.join(log_dir, "report.json"), "w", encoding="utf-8") as report_file:
    report_file.write(json.dumps(report))

ycruncher/ycruncher.py
LTTLabsOSS/markbench-tests
"""Collection of functions to assist in running of ycruncher test script"""
import os
from zipfile import ZipFile

import requests

YCRUNCHER_FOLDER_NAME = "y-cruncher v0.8.2.9522"
YCRUNCHER_ZIP_NAME = "y-cruncher v0.8.2.9522.zip"
SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))

def ycruncher_folder_exists() -> bool:
    """Check if ycruncher has been downloaded or not"""
    return os.path.isdir(os.path.join(SCRIPT_DIR, YCRUNCHER_FOLDER_NAME))


def download_ycruncher():
    """Download and extract Y-Cruncher"""
    download_url = "http://www.numberworld.org/y-cruncher/old_versions/y-cruncher%20v0.8.2.9522.zip"
    destination = os.path.join(SCRIPT_DIR, YCRUNCHER_ZIP_NAME)
    response = requests.get(download_url, allow_redirects=True, timeout=180)
    with open(destination, 'wb') as file:
        file.write(response.content)
    with ZipFile(destination, 'r') as zip_object:
        zip_object.extractall(path=SCRIPT_DIR)
        
ycruncher/ycruncher_utils.py
LTTLabsOSS/markbench-tests