Add Dota 2 harness (#74)

* Add initial Dota 2 harness

* update readme

* Update README

* Error in readme

* Please the linting gods

---------

Signed-off-by: nharris-lmg <105814489+nharris-lmg@users.noreply.github.com>

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 
 Changes are grouped by the date they are merged to the main branch of the repository and are ordered from newest to oldest. Dates use the ISO 8601 extended calendar date format, i.e. YYYY-MM-DD.
 
+## 2023-10-19
+
+- Add Dota 2 test harness.
+
 ## 2023-10-18
 
 - Add Blender barbershop render test harness.

cspell.json

@@ -13,6 +13,8 @@
     "Kikis",
     "Jcraft",
     "Kombustor",
+    "Dota",
+    "Keras",
     "twwh"
   ],
   "ignoreRegExpList": [

dota2/README.md

@@ -0,0 +1,20 @@
+# Dota 2
+Follows the benchmarking guide ["Benchmarking Dota 2" by JJ “PimpmuckL” Liebig](https://medium.com/layerth/benchmarking-dota-2-83c4322b12c0)
+
+## Prerequisites
+
+- Python 3.10+
+- Dota 2 installed
+- Keras OCR service
+
+## Options
+
+- `kerasHost`: string representing the IP address of the Keras service. e.x. `0.0.0.0` 
+- `kerasPort`: string representing the port of the Keras service. e.x. `8080`
+
+## Output
+
+report.json
+- `resolution`: string representing the resolution the test was run at, formatted as "[width]x[height]", e.x. `1920x1080`
+- `start_time`: number representing a timestamp of the test's start time in milliseconds
+- `end_time`: number representing a timestamp of the test's end time in milliseconds

dota2/dota2.py

@@ -0,0 +1,105 @@
+"""Dota 2 test script"""
+import logging
+import os
+import time
+import pydirectinput as user
+import sys
+from utils import console_command, get_resolution, copy_replay, copy_config, get_args
+
+sys.path.insert(1, os.path.join(sys.path[0], '..'))
+#pylint: disable=wrong-import-position
+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.keras_service import KerasService
+from harness_utils.steam import exec_steam_game
+
+
+SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
+LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
+PROCESS_NAME = "dota2.exe"
+STEAM_GAME_ID = 570
+
+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"))
+
+def start_game():
+    """Launch the game with console enabled and FPS unlocked"""
+    return exec_steam_game(STEAM_GAME_ID, game_params=["-console", "+fps_max 0"])
+
+def run_benchmark():
+    """Run dota2 benchmark"""
+    copy_replay()
+    copy_config()
+    setup_start_time = time.time()
+    start_game()
+    time.sleep(10)  # wait for game to load into main menu
+
+    # to skip logo screen
+    if kerasService.wait_for_word(word="va", timeout=20, interval=1):
+        logging.info('Game started. Entering main menu')
+        user.press("esc")
+        time.sleep(1)
+
+    # waiting about a minute for the main menu to appear
+    if kerasService.wait_for_word(word="heroes", timeout=80, interval=1) is None:
+        logging.error("Game didn't start in time. Check settings and try again.")
+        sys.exit(1)
+
+    logging.info('Starting benchmark')
+    user.press("\\")
+    time.sleep(0.2)
+    console_command("exec_async benchmark")
+    time.sleep(0.2)
+    user.press("\\")
+    time.sleep(5)
+
+    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)
+
+    # TODO -> Mark benchmark start time using video OCR by looking for a players name
+    if kerasService.wait_for_word(word="directed", timeout=100, interval=0.5) is None:
+        logging.error("Game didn't start in time. Check settings and try again.")
+        sys.exit(1)
+
+    time.sleep(100) # sleep duration during gameplay
+
+    if kerasService.wait_for_word(word="heroes", timeout=25, interval=1) is None:
+        logging.error("Main menu after running benchmark not found, exiting")
+        sys.exit(1)
+
+    logging.info("Run completed. Closing game.")
+    test_end_time = time.time()
+
+    elapsed_test_time = round((test_end_time - elapsed_setup_time), 2)
+    logging.info("Benchmark took %f seconds", elapsed_test_time)
+    terminate_processes(PROCESS_NAME)
+    return elapsed_setup_time, test_end_time
+
+try:
+    start_time, end_time = run_benchmark()
+    height, width = get_resolution()
+    report = {
+        "resolution": f"{width}x{height}",
+        "start_time": round((start_time * 1000)),
+        "end_time": round((end_time * 1000))
+    }
+
+    write_report_json(LOG_DIRECTORY, "report.json", report)
+except Exception as e:
+    logging.error("Something went wrong running the benchmark!")
+    logging.exception(e)
+    terminate_processes(PROCESS_NAME)
+    sys.exit(1)

dota2/manifest.yaml

@@ -0,0 +1,9 @@
+friendly_name: "DOTA 2"
+executable: "dota2.py"
+process_name: "dota2.exe"
+output_dir: "run"
+options:
+  - name: kerasHost
+    type: input
+  - name: kerasPort
+    type: input

dota2/utils.py

@@ -0,0 +1,129 @@
+"""Dota 2 test script utils"""
+from argparse import ArgumentParser
+import win32api
+import win32file
+import winreg
+import os
+import logging
+import re
+import shutil
+import pyautogui as gui
+import pydirectinput as user
+import sys
+from pathlib import Path
+
+sys.path.insert(1, os.path.join(sys.path[0], '..'))
+#pylint: disable=wrong-import-position
+
+USERNAME = os.getlogin()
+SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
+
+def console_command(command):
+    """Enter a console command"""
+    for char in command:
+        gui.press(char)
+    user.press("enter")
+
+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 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 install_location() -> any:
+    """Get installation location of Dota 2"""
+    reg_path = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\Steam App 570'
+    try:
+        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
+    #pylint: disable=undefined-variable
+    except WindowsError:
+        return None
+
+def copy_replay() -> None:
+    """Copy replay file to dota 2 folder"""
+    replay_path = os.path.join(install_location(), "game\\dota\\replays")
+    src_file = os.path.join(SCRIPT_DIRECTORY, "benchmark.dem")
+    destination_file = os.path.join(replay_path, os.path.basename(src_file))
+    if os.path.isfile(src_file) is not True:
+        source = r"\\Labs\labs\03_ProcessingFiles\Dota2\benchmark.dem"
+        root_dir = os.path.dirname(os.path.realpath(__file__))
+        destination = os.path.join(root_dir, "benchmark.dem")
+        shutil.copyfile(source, destination)
+    if not os.path.isfile(src_file):
+        raise Exception(f"Can't find no intro: {src_file}")
+    try:
+        Path(replay_path).mkdir(parents=True, exist_ok=True)
+    except FileExistsError as e:
+        logging.error(
+            "Could not create directory - likely due to non-directory file existing at path.")
+        raise e
+    logging.info("Copying: %s -> %s", src_file, destination_file)
+    shutil.copy(src_file, destination_file)
+
+def copy_config() -> None:
+    """Copy benchmark config to dota 2 folder"""
+    config_path = os.path.join(install_location(), "game\\dota\\cfg")
+    src_file = os.path.join(SCRIPT_DIRECTORY, "benchmark.cfg")
+    destination_file = os.path.join(config_path, os.path.basename(src_file))
+    if os.path.isfile(src_file) is not True:
+        source = r"\\Labs\labs\03_ProcessingFiles\Dota2\benchmark.cfg"
+        root_dir = os.path.dirname(os.path.realpath(__file__))
+        destination = os.path.join(root_dir, "benchmark.cfg")
+        shutil.copyfile(source, destination)
+    if not os.path.isfile(src_file):
+        raise Exception(f"Can't find no config: {src_file}")
+    try:
+        Path(config_path).mkdir(parents=True, exist_ok=True)
+    except FileExistsError as e:
+        logging.error(
+            "Could not create directory - likely due to non-directory file existing at path.")
+        raise e
+    logging.info("Copying: %s -> %s", src_file, destination_file)
+    shutil.copy(src_file, destination_file)
+
+def get_resolution():
+    """Get current resolution from settings file"""
+    video_config = os.path.join(install_location(), "game\\dota\\cfg\\video.txt")
+    height_pattern = re.compile(r"\"setting.defaultresheight\"		\"(\d+)\"")
+    width_pattern = re.compile(r"\"setting.defaultres\"		\"(\d+)\"")
+    cfg = f"{video_config}"
+    height = 0
+    width = 0
+    with open(cfg, encoding="utf-8") as f:
+        lines = f.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)
+            if height != 0 and width !=0:
+                return (height, width)
+    return (height, width)

pylintrc

@@ -4,4 +4,7 @@ disable=
     wrong-import-order,
     wrong-import-position,
     broad-exception-caught,
-    line-too-long
+    broad-exception-raised,
+    line-too-long,
+    fixme,
+    c-extension-no-member

recording_session/manifest.yaml

@@ -1,4 +1,5 @@
 friendly_name: "Recording Session"
 executable: "dummy.py"
+output_dir: "run"