Jl harness (#136)

- adds logging to output.py in harness utils, to remove the need to
write the logging function for every harness
- writes a more convenient function to call keras and parse keras args
within misc.py in harness_utils, does not remove any functionality from
other harnesses
- added function to call time function while rounding to int 
- added default paus value for press n times
- dota2: changing to path instead of os
- ac shadows: os to path conversion, uses imported functions instead of
writing
- general linting 
- the last of us part 1 lint
- new the last of us part 2 harness
This commit is contained in:
j-lin-lmg
2025-05-13 14:53:25 -07:00
committed by GitHub
parent 1e2abd47b3
commit 6f47569db6
7 changed files with 405 additions and 91 deletions

View File

@@ -1,36 +1,36 @@
# pylint: disable=missing-module-docstring
from argparse import ArgumentParser
import logging
import os
from pathlib import Path
import time
import sys
import re
import pydirectinput as user
import getpass
sys.path.insert(1, os.path.join(sys.path[0], '..'))
sys.path.insert(1, str(Path(sys.path[0]).parent))
# pylint: disable=wrong-import-position
from harness_utils.process import terminate_processes
from harness_utils.output import (
format_resolution,
setup_log_directory,
setup_logging,
write_report_json,
seconds_to_milliseconds,
DEFAULT_LOGGING_FORMAT,
DEFAULT_DATE_FORMAT
)
from harness_utils.steam import get_build_id, exec_steam_game
from harness_utils.keras_service import KerasService
from harness_utils.artifacts import ArtifactManager, ArtifactType
from harness_utils.misc import press_n_times
USERNAME = getpass.getuser()
from harness_utils.misc import (
press_n_times,
int_time,
find_word,
keras_args)
SCRIPT_DIR = Path(__file__).resolve().parent
LOG_DIR = SCRIPT_DIR.joinpath("run")
PROCESS_NAME = "ACShadows.exe"
USERNAME = getpass.getuser()
STEAM_GAME_ID = 3159330
SCRIPT_DIR = Path(__file__).resolve().parent
LOG_DIR = SCRIPT_DIR / "run"
PROCESS_NAME = "ACShadows.exe"
CONFIG_LOCATION = f"C:\\Users\\{USERNAME}\\Documents\\Assassin's Creed Shadows"
CONFIG_FILENAME = "ACShadows.ini"
@@ -56,40 +56,30 @@ def read_current_resolution():
return (height_value, width_value)
def find_word(keras_service, word, msg, timeout=30, interval=1):
"""function to call keras """
if keras_service.wait_for_word(
word=word, timeout=timeout, interval=interval) is None:
logging.info(msg)
sys.exit(1)
def int_time():
"""rounds time to int"""
return int(time.time())
def delete_videos():
"""deletes intro videos"""
base_dir = r"C:\Program Files (x86)\Steam\steamapps\common\Assassin's Creed Shadows"
videos_dir = os.path.join(base_dir, "videos")
videos_en_dir = os.path.join(videos_dir, "en")
base_dir = Path(
r"C:\Program Files (x86)\Steam\steamapps\common\Assassin's Creed Shadows")
videos_dir = base_dir / "videos"
videos_en_dir = videos_dir / "en"
# List of video files to delete
videos_to_delete = [
os.path.join(videos_dir, "ANVIL_Logo.webm"),
os.path.join(videos_dir, "INTEL_Logo.webm"),
os.path.join(videos_dir, "HUB_Bootflow_FranchiseIntro.webm"),
os.path.join(videos_dir, "UbisoftLogo.webm"),
os.path.join(videos_en_dir, "Epilepsy.webm"),
os.path.join(videos_en_dir, "warning_disclaimer.webm"),
os.path.join(videos_en_dir, "WarningSaving.webm")
videos_dir / "ANVIL_Logo.webm",
videos_dir / "INTEL_Logo.webm",
videos_dir / "HUB_Bootflow_FranchiseIntro.webm",
videos_dir / "HUB_Bootflow_AbstergoIntro.webm",
videos_dir / "UbisoftLogo.webm",
videos_en_dir / "Epilepsy.webm",
videos_en_dir / "warning_disclaimer.webm",
videos_en_dir / "WarningSaving.webm"
]
for file_path in videos_to_delete:
if os.path.exists(file_path):
if file_path.exists():
try:
os.remove(file_path)
file_path.unlink()
logging.info("Deleted: %s", file_path)
except Exception as e:
logging.error("Error deleting %s: %s", file_path, e)
@@ -97,15 +87,15 @@ def delete_videos():
def move_benchmark_file():
"""moves html benchmark results to log folder"""
src_dir = f"C:\\Users\\{USERNAME}\\Documents\\Assassin's Creed Shadows\\benchmark_reports"
src_dir = Path(
f"C:\\Users\\{USERNAME}\\Documents\\Assassin's Creed Shadows\\benchmark_reports")
for filename in os.listdir(src_dir):
src_path = os.path.join(src_dir, filename)
dest_path = os.path.join(LOG_DIR, filename)
for src_path in src_dir.iterdir():
dest_path = LOG_DIR / src_path.name
if os.path.isfile(src_path):
if src_path.is_file():
try:
os.rename(src_path, dest_path)
src_path.rename(dest_path)
logging.info("Benchmark HTML moved")
except Exception as e:
logging.error("Failed to move %s: %s", src_path, e)
@@ -190,7 +180,7 @@ def run_benchmark(keras_service):
user.press("f1")
find_word(keras_service, "system", "couldn't find system")
find_word(keras_service, "system", "Couldn't find 'System' button")
user.press("down")
@@ -198,10 +188,16 @@ def run_benchmark(keras_service):
user.press("space")
find_word(keras_service, "benchmark", "couldn't find benchmark")
find_word(
keras_service, "benchmark",
"couldn't find 'benchmark' on screen before settings")
navi_settings(am)
find_word(
keras_service, "benchmark",
"couldn't find 'benchmark' on screen after settings")
user.press("down")
time.sleep(1)
@@ -221,10 +217,7 @@ def run_benchmark(keras_service):
time.sleep(100)
if keras_service.wait_for_word(
word="results", timeout=30, interval=1) is None:
logging.info("did not find end screen")
sys.exit(1)
find_word(keras_service, "results", "did not find results screen", 60)
test_end_time = int_time() - 2
@@ -252,28 +245,10 @@ def run_benchmark(keras_service):
return test_start_time, test_end_time
def setup_logging():
"""setup logging"""
setup_log_directory(LOG_DIR)
logging.basicConfig(filename=f'{LOG_DIR}/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)
def main():
"""entry point"""
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()
keras_service = KerasService(args.keras_host, args.keras_port)
keras_service = KerasService(
keras_args().keras_host, keras_args().keras_port)
start_time, endtime = run_benchmark(keras_service)
height, width = read_current_resolution()
report = {
@@ -287,7 +262,7 @@ def main():
if __name__ == "__main__":
try:
setup_logging()
setup_logging(LOG_DIR)
main()
except Exception as ex:
logging.error("Something went wrong running the benchmark!")

View File

@@ -1,13 +1,13 @@
"""Dota 2 test script"""
import logging
import os
from pathlib import Path
import time
import pyautogui as gui
import pydirectinput as user
import sys
from dota2_utils import get_resolution, copy_replay, copy_config, get_args
sys.path.insert(1, os.path.join(sys.path[0], '..'))
sys.path.insert(1, str(Path(sys.path[0]).parent))
from harness_utils.output import (
setup_log_directory,
@@ -22,8 +22,8 @@ from harness_utils.steam import exec_steam_game
from harness_utils.artifacts import ArtifactManager, ArtifactType
SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
LOG_DIRECTORY = os.path.join(SCRIPT_DIRECTORY, "run")
SCRIPT_DIRECTORY = Path(__file__).resolve().parent
LOG_DIRECTORY = SCRIPT_DIRECTORY / "run"
PROCESS_NAME = "dota2.exe"
STEAM_GAME_ID = 570
@@ -198,9 +198,9 @@ def run_benchmark():
try:
start_time, end_time = run_benchmark()
height, width = get_resolution()
res_height, res_width = get_resolution()
report = {
"resolution": format_resolution(width, height),
"resolution": format_resolution(res_width, res_height),
"start_time": seconds_to_milliseconds(start_time),
"end_time": seconds_to_milliseconds(end_time)
}

View File

@@ -1,4 +1,5 @@
"""Misc utility functions"""
from argparse import ArgumentParser
import logging
import os
from pathlib import Path
@@ -10,6 +11,8 @@ import requests
import vgamepad as vg
import json
import re
import sys
class LTTGamePad360(vg.VX360Gamepad):
"""
@@ -19,7 +22,8 @@ class LTTGamePad360(vg.VX360Gamepad):
This class extension provides some useful functions to make your code look a little cleaner when
implemented in our harnesses.
"""
def single_press(self, button = vg.XUSB_BUTTON.XUSB_GAMEPAD_DPAD_DOWN, pause = 0.1):
def single_press(self, button=vg.XUSB_BUTTON.XUSB_GAMEPAD_DPAD_DOWN, pause=0.1):
"""
Custom function to perform a single press of a specified gamepad button
@@ -59,6 +63,7 @@ class LTTGamePad360(vg.VX360Gamepad):
self.single_press(button)
time.sleep(pause)
class LTTGamePadDS4(vg.VDS4Gamepad):
"""
Class extension for the virtual game pad library
@@ -67,7 +72,8 @@ class LTTGamePadDS4(vg.VDS4Gamepad):
This class extension provides some useful functions to make your code look a little cleaner when
implemented in our harnesses.
"""
def single_button_press(self, button = vg.DS4_BUTTONS.DS4_BUTTON_CROSS, fastpause = 0.05):
def single_button_press(self, button=vg.DS4_BUTTONS.DS4_BUTTON_CROSS, fastpause=0.05):
"""
Custom function to perform a single press of a specified gamepad digital button
@@ -96,7 +102,6 @@ class LTTGamePadDS4(vg.VDS4Gamepad):
self.release_button(button=button)
self.update()
def button_press_n_times(self, button: vg.DS4_BUTTONS, n: int, pause: float):
"""
Sometimes we need to press a certain gamepad button multiple times in a row, this loop does that for you
@@ -105,7 +110,7 @@ class LTTGamePadDS4(vg.VDS4Gamepad):
self.single_button_press(button)
time.sleep(pause)
def single_dpad_press(self, direction = vg.DS4_DPAD_DIRECTIONS.DS4_BUTTON_DPAD_SOUTH, pause = 0.1):
def single_dpad_press(self, direction=vg.DS4_DPAD_DIRECTIONS.DS4_BUTTON_DPAD_SOUTH, pause=0.1):
"""
Custom function to perform a single press of a specified gamepad button
@@ -139,6 +144,7 @@ class LTTGamePadDS4(vg.VDS4Gamepad):
self.single_dpad_press(direction)
time.sleep(pause)
def clickme(x: int, y: int):
"""Pyautogui's click function sucks, this should do the trick"""
gui.moveTo(x, y)
@@ -147,6 +153,7 @@ def clickme(x: int, y: int):
time.sleep(0.2)
gui.mouseUp()
def mouse_scroll_n_times(n: int, scroll_amount: int, pause: float):
"""
Pyautogui's mouse scroll function often fails to actually scroll in game menus, this functions solves that problem
@@ -159,12 +166,19 @@ def mouse_scroll_n_times(n: int, scroll_amount: int, pause: float):
gui.vscroll(scroll_amount)
time.sleep(pause)
def press_n_times(key: str, n: int, pause: float):
def int_time() -> int:
"""Returns the current time in seconds since epoch as an integer"""
return int(time.time())
def press_n_times(key: str, n: int, pause: float = 0.5):
"""A helper function press the same button multiple times"""
for _ in range(n):
user.press(key)
time.sleep(pause)
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.
@@ -228,3 +242,20 @@ def find_eg_game_version(gamefoldername: str) -> str:
print(f"Error: {e}")
return None
def find_word(keras_service, word, msg, timeout=30, interval=1):
"""Function to call Keras service to find a word in the screen"""
if keras_service.wait_for_word(word=word, timeout=timeout, interval=interval) is None:
logging.error(msg)
sys.exit(1)
def keras_args():
"""helper function to get args for keras"""
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()

View File

@@ -1,10 +1,12 @@
"""Functions related to logging and formatting output from test harnesses."""
import json
import os
import logging
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):
@@ -25,3 +27,17 @@ def format_resolution(width: int, height: int) -> str:
def seconds_to_milliseconds(seconds: float | int) -> int:
"""Convert seconds to milliseconds"""
return round((seconds * 1000))
def setup_logging(log_directory: str) -> None:
"""Sets up logging for the harness"""
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)

View File

@@ -34,19 +34,20 @@ PROCESS_NAME = "tlou"
user.FAILSAFE = False
def take_screenshots(am: ArtifactManager) -> None:
"""Take screenshots of the benchmark settings"""
logging.info("Taking screenshots of benchmark settings")
press_n_times("s",2,0.2 )
press_n_times("s", 2, 0.2)
user.press("enter")
press_n_times("s",4,0.2 )
press_n_times("s", 4, 0.2)
user.press("enter")
am.take_screenshot("video1.png", ArtifactType.CONFIG_IMAGE, "screenshot of video settings1")
press_n_times("s",15,0.2)
press_n_times("s", 15, 0.2)
am.take_screenshot("video2.png", ArtifactType.CONFIG_IMAGE, "screenshot of video settings2")
press_n_times("s",6, 0.2)
press_n_times("s", 6, 0.2)
am.take_screenshot("video3.png", ArtifactType.CONFIG_IMAGE, "screenshot of video settings3")
user.press("backspace")
@@ -69,6 +70,7 @@ def take_screenshots(am: ArtifactManager) -> None:
user.press("backspace")
press_n_times("w", 2, 0.2)
def navigate_main_menu(am: ArtifactManager) -> None:
"""Input to navigate main menu"""
logging.info("Navigating main menu")
@@ -174,7 +176,7 @@ 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",
os.environ["HOMEPATH"], "Saved Games", "The Last of Us Part I",
"users", str(steam_id), "screeninfo.cfg"
)
height, width = get_resolution(config_path)
@@ -183,8 +185,8 @@ try:
"start_time": seconds_to_milliseconds(start_time),
"end_time": seconds_to_milliseconds(end_time)
}
write_report_json(LOG_DIRECTORY, "report.json", report)
except Exception as e:
logging.error("Something went wrong running the benchmark!")
logging.exception(e)

View File

@@ -0,0 +1,9 @@
friendly_name: "The Last of Us Part II"
executable: "tlou2.py"
process_name: "tlou-ii.exe"
output_dir: "run"
options:
- name: kerasHost
type: input
- name: kerasPort
type: input

View File

@@ -0,0 +1,281 @@
"""The Last of Us Part I test script"""
import logging
from pathlib import Path
import time
import sys
import pydirectinput as user
import getpass
import winreg # for accessing settings, including resolution, in the registry
import shutil
sys.path.insert(1, str(Path(sys.path[0]).parent))
from harness_utils.keras_service import KerasService
from harness_utils.output import (
format_resolution,
seconds_to_milliseconds,
write_report_json,
setup_logging,
)
from harness_utils.process import terminate_processes
from harness_utils.steam import (
exec_steam_run_command,
)
from harness_utils.artifacts import ArtifactManager, ArtifactType
from harness_utils.misc import (
int_time,
find_word,
press_n_times,
keras_args)
USERNAME = getpass.getuser()
STEAM_GAME_ID = 2531310
SCRIPT_DIRECTORY = Path(__file__).resolve().parent
LOG_DIRECTORY = SCRIPT_DIRECTORY / "run"
PROCESS_NAME = "tlou-ii.exe"
user.FAILSAFE = False
def reset_savedata():
"""
Deletes the savegame folder from the local directory and replaces it with a new one from the network drive.
"""
local_savegame_path = Path(
f"C:\\Users\\{USERNAME}\\Documents\\The Last of Us Part II\\76561199405246658\\savedata") # make this global
network_savegame_path = Path(r"\\Labs\Labs\03_ProcessingFiles\The Last of Us Part II\savedata")
# Delete the local savedata folder if it exists
if local_savegame_path.exists() and local_savegame_path.is_dir():
shutil.rmtree(local_savegame_path)
logging.info("Deleted local savedata folder: %s", local_savegame_path)
# Copy the savedata folder from the network drive
try:
shutil.copytree(network_savegame_path, local_savegame_path)
logging.info("Copied savedata folder from %s to %s", network_savegame_path, local_savegame_path)
except Exception as e:
logging.error("Failed to copy savedata folder: %s", e)
# Check if the newly copied directory contains a folder called SAVEFILE0A
def delete_autosave():
"""
Deletes the autosave folder from the local directory if it exists.
"""
local_savegame_path = Path(f"C:\\Users\\{USERNAME}\\Documents\\The Last of Us Part II\\76561199405246658\\savedata")
savefile_path = local_savegame_path / "SAVEFILE0A" # check for autosaved file, delete if exists
if savefile_path.exists() and savefile_path.is_dir():
shutil.rmtree(savefile_path)
logging.info("Deleted folder: %s", savefile_path)
def get_current_resolution():
"""
Returns:
tuple: (width, height)
Reads resolutions settings from registry
"""
key_path = r"Software\Naughty Dog\The Last of Us Part II\Graphics"
fullscreen_width = read_registry_value(key_path, "FullscreenWidth")
fullscreen_height = read_registry_value(key_path, "FullscreenHeight")
return (fullscreen_width, fullscreen_height)
def read_registry_value(key_path, value_name):
"""
Reads value from registry
A helper function for get_current_resolution
"""
try:
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key:
value, _ = winreg.QueryValueEx(key, value_name)
return value
except FileNotFoundError:
logging.error("Registry key not found: %s", value_name)
return None
except OSError as e:
logging.error("Error reading registry value: %s", e)
return None
def run_benchmark(keras_service: KerasService) -> tuple:
"""Starts Game, Sets Settings, and Runs Benchmark"""
exec_steam_run_command(STEAM_GAME_ID)
setup_start_time = int_time()
am = ArtifactManager(LOG_DIRECTORY)
if keras_service.wait_for_word(word="sony", timeout=60, interval=0.2) is None:
logging.error("Couldn't find 'sony'")
else:
user.press("escape")
find_word(keras_service, "story", "Couldn't find main menu : 'story'")
press_n_times("down", 2)
# navigate settings
navigate_settings(am, keras_service)
find_word(keras_service, "story", "Couldn't find main menu the second time : 'story'")
press_n_times("up", 2)
user.press("space")
time.sleep(0.3)
if keras_service.wait_for_word(word="continue", timeout=5, interval=0.2) is None:
user.press("down")
else:
press_n_times("down", 2)
delete_autosave()
time.sleep(0.3)
user.press("space")
time.sleep(0.3)
if keras_service.wait_for_word(word="autosave", timeout=5, interval=0.2) is None:
user.press("space")
else:
user.press("up")
time.sleep(0.3)
user.press("space")
time.sleep(0.3)
user.press("left")
time.sleep(0.3)
user.press("space")
setup_end_time = test_start_time = test_end_time = int_time()
elapsed_setup_time = setup_end_time - setup_start_time
logging.info("Setup took %f seconds", elapsed_setup_time)
# time of benchmark usually is 4:23 = 263 seconds
if keras_service.wait_for_word(word="man", timeout=100, interval=0.2) is not None:
test_start_time = int_time() - 14
time.sleep(240)
else:
logging.error("couldn't find 'man'")
time.sleep(150)
if keras_service.wait_for_word(word="rush", timeout=100, interval=0.2) is not None:
time.sleep(3)
test_end_time = int_time()
else:
logging.error("couldn't find 'rush', marks end of benchmark")
test_end_time = int_time()
elapsed_test_time = test_end_time - test_start_time
logging.info("Test took %f seconds", elapsed_test_time)
terminate_processes(PROCESS_NAME)
am.create_manifest()
return test_start_time, test_end_time
def navigate_settings(am: ArtifactManager, keras: KerasService) -> None:
"""Navigate through settings and take screenshots.
Exits to main menu after taking screenshots.
"""
user.press("space")
find_word(keras, "display", "Couldn't find display")
time.sleep(5) # slow cards may miss the first down
press_n_times("down", 4)
user.press("space")
time.sleep(0.5)
find_word(keras, "resolution", "Couldn't find resolution")
am.take_screenshot("display1.png", ArtifactType.CONFIG_IMAGE, "display settings 1")
user.press("up")
find_word(keras, "brightness", "Couldn't find brightness")
am.take_screenshot("display2.png", ArtifactType.CONFIG_IMAGE, "display settings 2")
user.press("q") # swaps to graphics settings
time.sleep(0.5)
find_word(keras, "preset", "Couldn't find preset")
am.take_screenshot("graphics1.png", ArtifactType.CONFIG_IMAGE, "graphics settings 1")
user.press("up")
find_word(keras, "dirt", "Couldn't find dirt")
am.take_screenshot("graphics3.png", ArtifactType.CONFIG_IMAGE,
"graphics settings 3") # is at the bottom of the menu
press_n_times("up", 12)
find_word(keras, "scattering", "Couldn't find scattering")
am.take_screenshot("graphics2.png", ArtifactType.CONFIG_IMAGE, "graphics settings 2")
press_n_times("escape", 2)
def main():
"""Main function to run the benchmark"""
try:
logging.info("Starting The Last of Us Part II benchmark")
keras_service = KerasService(keras_args().keras_host, keras_args().keras_port)
reset_savedata()
start_time, end_time = run_benchmark(keras_service)
resolution_tuple = get_current_resolution()
report = {
"resolution": format_resolution(resolution_tuple[0], resolution_tuple[1]),
"start_time": seconds_to_milliseconds(start_time), # secconds to miliseconds
"end_time": seconds_to_milliseconds(end_time),
}
write_report_json(LOG_DIRECTORY, "report.json", report)
except Exception as e:
logging.error("An error occurred: %s", e)
logging.exception(e)
terminate_processes(PROCESS_NAME)
sys.exit(1)
if __name__ == "__main__":
setup_logging(LOG_DIRECTORY)
main()