Audit F1 23 (#7)

* fix f1_22 end time

* pep8 lint utils

* Add readme

* pep8 lint

* Add gitignore with vscode assets

* remove untracked files

.gitignore

@@ -0,0 +1,12 @@
+.vscode/*
+!.vscode/settings.json
+!.vscode/tasks.json
+!.vscode/launch.json
+!.vscode/extensions.json
+!.vscode/*.code-snippets
+
+# Local History for Visual Studio Code
+.history/
+
+# Built Visual Studio Code Extensions
+*.vsix

.vscode/settings.json

@@ -1,5 +0,0 @@
-{
-  "[python]": {
-    "editor.defaultFormatter": "ms-python.black-formatter"
-  },
-}

F1_22/f1.py

@@ -183,10 +183,12 @@ def run_benchmark():
         )
         sys.exit(1)
 
-    elapsed_test_time = round(time.time() - test_start_time, 2)
+    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, elapsed_test_time
+    return test_start_time, test_end_time
 
 
 setup_log_directory(LOG_DIRECTORY)

F1_23/README.md

@@ -0,0 +1,18 @@
+# F1 23
+
+## Prerequisites
+
+- Python 3.10+
+- F1 23 installed
+
+## 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 serivce. e.x. `8080`
+
+## Output
+
+report.json
+- `resolution`: string representing the resolution the test was run at, formatted as "[width]x[heigt]", 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

F1_23/f1_23.py

@@ -1,20 +1,30 @@
-from argparse import ArgumentParser
+"""F1 23 test script"""
 import logging
+from argparse import ArgumentParser
 import os.path
 import time
-import pydirectinput as user
 from subprocess import Popen
 import sys
-
+import pydirectinput as user
-
 from f1_23_utils import get_resolution, remove_intro_videos, skippable
 
-sys.path.insert(1, os.path.join(sys.path[0], '..'))
+sys.path.insert(1, os.path.join(sys.path[0], ".."))
+
+# pylint: disable=wrong-import-position
 
-from harness_utils.logging import *
-from harness_utils.process import terminate_processes
-from harness_utils.keras_service import KerasService
 from harness_utils.steam import DEFAULT_EXECUTABLE_PATH as STEAM_PATH
+from harness_utils.keras_service import KerasService
+from harness_utils.process import terminate_processes
+from harness_utils.logging 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")
@@ -25,66 +35,118 @@ PROCESS_NAME = "F1_23"
 
 
 def is_word_on_screen(word: str, attempts: int = 5, delay_seconds: int = 1) -> bool:
+    """Sends screenshot to Keras service to search for a given word a specified number of times"""
     for _ in range(attempts):
         result = kerasService.capture_screenshot_find_word(word)
-        if result != None:
+        if result is not None:
             return True
         time.sleep(delay_seconds)
     return False
 
 
 def start_game():
+    """Starts the game via steam command."""
     steam_run_arg = "steam://rungameid/" + str(STEAM_GAME_ID)
     start_cmd = STEAM_PATH + "\\" + STEAM_EXECUTABLE
-    logging.info(start_cmd + " " + steam_run_arg)
+    logging.info("%s %s", start_cmd, steam_run_arg)
-    thread =  Popen([start_cmd, steam_run_arg])
+    thread = Popen([start_cmd, steam_run_arg])
     return thread
 
+
 def official() -> any:
+    """Look for product"""
     return is_word_on_screen(word="product", attempts=20, delay_seconds=0.5)
-    
+
+
 def press() -> any:
+    """Look for press"""
     return is_word_on_screen(word="press", attempts=40, delay_seconds=2)
 
+
 def login() -> any:
+    """Look for login"""
     return is_word_on_screen(word="login", attempts=100, delay_seconds=0.2)
 
+
 def benchmark_start() -> any:
+    """Look for lap"""
     return is_word_on_screen(word="lap", attempts=15, delay_seconds=3)
 
+
 def results() -> any:
+    """Look for results"""
     return is_word_on_screen(word="results", attempts=20, delay_seconds=3)
 
+
 def menu() -> any:
+    """Look for theatre"""
     return is_word_on_screen(word="theatre", attempts=5, delay_seconds=3)
 
-def settings() -> any:
-    return is_word_on_screen(word="settings", attempts=5, delay_seconds=3)
 
-def graphics() -> any:
+def find_settings() -> any:
-    return is_word_on_screen(word="graphics", attempts=5, delay_seconds=3)
+    """Look for and enter settings"""
-
+    if not is_word_on_screen(word="settings", attempts=5, delay_seconds=3):
-def benchmark() -> any:
+        logging.info("Didn't find settings!")
-    return is_word_on_screen(word="benchmark", attempts=5, delay_seconds=3)
+        sys.exit(1)
-
+    user.press("enter")
-def weather() -> any:
+    time.sleep(1.5)
-    return is_word_on_screen(word="weather", attempts=5, delay_seconds=3)
 
 
-# ASSUMPTIONS
+def find_graphics() -> any:
-# 1. We are doing 3 laps
+    """Look for and enter graphics settings"""
-def run_benchmark():
+    if not is_word_on_screen(word="graphics", attempts=5, delay_seconds=3):
-    remove_intro_videos(skippable)
+        logging.info("Didn't find graphics!")
-    start_game()
+        sys.exit(1)
-    t1 = time.time()
+    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 is_word_on_screen(word="benchmark", attempts=5, delay_seconds=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 is_word_on_screen(word="weather", attempts=5, delay_seconds=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)
 
-    # press space through the warnings
+
-    start_game_screen= time.time()
+def navigate_startup():
-    while (True):
+    """press space through the warnings and navigate startup menus"""
+    start_game_screen = time.time()
+    while True:
         if official():
-            logging.info('Game started. Skipping intros.')
+            logging.info("Game started. Skipping intros.")
             user.press("space")
             time.sleep(1)
             user.press("space")
@@ -92,39 +154,41 @@ def run_benchmark():
             user.press("space")
             time.sleep(4)
             break
-        elif time.time()-start_game_screen>60:
+        if time.time() - start_game_screen > 60:
-            logging.info("Game didn't start in time. Check settings and try again.")
+            logging.info(
-            exit(1)
+                "Game didn't start in time. Check settings and try again.")
+            sys.exit(1)
         logging.info("Game hasn't started. Trying again in 5 seconds")
         time.sleep(5)
-    
+
     # Press enter to proceed to the main menu
-    start_game_screen= time.time()
+    start_game_screen = time.time()
-    while (True):
+    while True:
         if press():
-            logging.info('Hit the title screen. Continuing')
+            logging.info("Hit the title screen. Continuing")
             user.press("enter")
             time.sleep(1)
             break
-        elif time.time()-start_game_screen>60:
+        if time.time() - start_game_screen > 60:
-            logging.info("Game didn't start in time. Check settings and try again.")
+            logging.info(
-            exit(1)
+                "Game didn't start in time. Check settings and try again.")
+            sys.exit(1)
         logging.info("Game hasn't started. Trying again in 5 seconds")
         time.sleep(5)
 
     # cancel logging into ea services
-    if login():   
+    if login():
-        logging.info('Cancelling logging in.')
+        logging.info("Cancelling logging in.")
         user.press("enter")
         time.sleep(2)
 
-    ###
+
-    # SHOULD BE ON MAIN MENU RIGHT NOW
+def navigate_menu():
-    ###
+    """Simulate inputs to navigate to benchmark option."""
     menu_screen = time.time()
-    while (True):
+    while True:
         if menu():
-            logging.info('Saw the options! we are good to go!')
+            logging.info("Saw the options! we are good to go!")
             time.sleep(1)
             user.press("down")
             time.sleep(0.5)
@@ -143,124 +207,107 @@ def run_benchmark():
             user.press("enter")
             time.sleep(2)
             break
-        elif time.time()-menu_screen>60:
+        if time.time() - menu_screen > 60:
             logging.info("Didn't land on the main menu!")
-            exit(1)
+            sys.exit(1)
         logging.info("Game still loading. Trying again in 10 seconds")
         time.sleep(10)
 
-    # Enter settings
+    find_settings()
-    if settings():
+    find_graphics()
-        user.press("enter")
+    find_benchmark()
-        time.sleep(1.5)
+    find_weather() # Run benchmark!
 
-    # Enter graphics settings
+def run_benchmark():
-    if graphics():
+    """Runs the actual benchmark."""
-        user.press("right")
+    remove_intro_videos(skippable)
-        time.sleep(0.5)
+    start_game()
-        user.press("enter")
+    setup_start_time = time.time()
-        time.sleep(1.5)
+    time.sleep(2)
+    navigate_startup()
+    navigate_menu()
 
-    # Enter benchmark options
+    loading_screen_start = time.time()
-    if benchmark():
+    while True:
-        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)
-
-    # Run benchmark!
-    if weather():
-        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)
-
-    loading_screen_start= time.time()
-    while (True):
         if benchmark_start():
-            start_time = time.time()
+            test_start_time = time.time()
-            logging.info("Benchmark started. Waiting for benchmark to complete.")
+            logging.info(
+                "Benchmark started. Waiting for benchmark to complete.")
             break
-        elif time.time()-loading_screen_start>60:
+        if time.time() - loading_screen_start > 60:
             logging.info("Benchmark didn't start.")
-            exit(1)
+            sys.exit(1)
         logging.info("Benchmark hasn't started. Trying again in 10 seconds")
         time.sleep(10)
 
-    t2 = time.time()
+    elapsed_setup_time = round(time.time() - setup_start_time, 2)
-    logging.info(f"Setup took {round((t2 - t1), 2)} seconds")
+    logging.info("Setup took %f seconds", elapsed_setup_time)
 
     # sleep for 3 laps
     time.sleep(350)
 
-    results_screen_start= time.time()
+    results_screen_start = time.time()
-    while (True):
+    while True:
         if results():
             logging.info("Results screen was found! Finishing benchmark.")
             break
-        elif time.time()-results_screen_start>60:
+        if time.time() - results_screen_start > 60:
-            logging.info("Results screen was not found! Did harness not wait long enough? Or test was too long?")
+            logging.info("Results screen was not found!" +
-            exit(1)
+                "Did harness not wait long enough? Or test was too long?")
+            sys.exit(1)
         logging.info("Benchmark hasn't finished. Trying again in 10 seconds")
         time.sleep(10)
 
-    end_time = time.time()
+    test_end_time = time.time()
-    logging.info(f"Benchmark took {round((end_time - start_time), 2)} seconds")
+    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 start_time, end_time
+    return test_start_time, test_end_time
 
 
 setup_log_directory(LOG_DIRECTORY)
 
-logging.basicConfig(filename=f'{LOG_DIRECTORY}/harness.log',
+logging.basicConfig(
-                    format=DEFAULT_LOGGING_FORMAT,
+    filename=f"{LOG_DIRECTORY}/harness.log",
-                    datefmt=DEFAULT_DATE_FORMAT,
+    format=DEFAULT_LOGGING_FORMAT,
-                    level=logging.DEBUG)
+    datefmt=DEFAULT_DATE_FORMAT,
+    level=logging.DEBUG,
+)
 console = logging.StreamHandler()
 formatter = logging.Formatter(DEFAULT_LOGGING_FORMAT)
 console.setFormatter(formatter)
-logging.getLogger('').addHandler(console)
+logging.getLogger("").addHandler(console)
 
 parser = ArgumentParser()
-parser.add_argument("--kerasHost", dest="keras_host", help="Host for Keras OCR service", required=True)
+parser.add_argument(
-parser.add_argument("--kerasPort", dest="keras_port", help="Port for Keras OCR service", required=True)
+    "--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(
+kerasService = KerasService(
-    LOG_DIRECTORY, "screenshot.jpg"))
+    args.keras_host, args.keras_port, os.path.join(
+        LOG_DIRECTORY, "screenshot.jpg")
+)
 
 try:
     start_time, end_time = run_benchmark()
     width, height = get_resolution()
-    result = {
+    report = {
         "resolution": format_resolution(width, height),
-        "graphics_preset": 'current',
+        "graphics_preset": "current",
         "start_time": seconds_to_milliseconds(start_time),
-        "end_time": seconds_to_milliseconds(end_time)
+        "end_time": seconds_to_milliseconds(end_time),
     }
 
-    write_report_json(LOG_DIRECTORY, "report.json", result)
+    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)
-    exit(1)
+    sys.exit(1)

F1_23/f1_23_utils.py

@@ -1,19 +1,20 @@
+"""Utility functions supporting F1 23 test script."""
 import os
 import re
 import winreg
 import logging
 
-# Stub
+def get_resolution() -> tuple[int]:   
-def get_resolution():    
+    """Gets resolution width and height from local xml file created by game."""
-    USERNAME = os.getlogin()
+    username = os.getlogin()
-    CONFIG_LOCATION = f"C:\\Users\\{USERNAME}\\Documents\\My Games\\F1 23\\hardwaresettings"
+    config_path = f"C:\\Users\\{username}\\Documents\\My Games\\F1 22\\hardwaresettings"
-    CONFIG_FILENAME = "hardware_settings_config.xml"
+    config_filename = "hardware_settings_config.xml"
     resolution = re.compile(r"<resolution width=\"(\d+)\" height=\"(\d+)\"")
-    cfg = f"{CONFIG_LOCATION}\\{CONFIG_FILENAME}"
+    cfg = f"{config_path}\\{config_filename}"
     height = 0
     width = 0
-    with open(cfg) as f:
+    with open(cfg, encoding="utf-8") as file:
-        lines = f.readlines()
+        lines = file.readlines()
         for line in lines:
             height_match = resolution.search(line)
             width_match = resolution.search(line)
@@ -23,18 +24,20 @@ def get_resolution():
                 width = width_match.group(1)
     return (width, height)
 
+
 def remove_intro_videos(file_paths: list[str]) -> None:
+    """Remove video files from paths to speed up game startup."""
     for video in file_paths:
         try:
             os.remove(video)
-            logging.info(f"Removing video {video}")
+            logging.info("Removing video %s", video)
         except FileNotFoundError:
-            logging.info(f"Video already removed {video}")
             # If file not found, it has likely already been deleted before.
-            pass
+            logging.info("Video already removed %s", video)
-        
 
-def F1_23_DIRECTORY() -> any:
+
+def f1_23_directory() -> any:
+    """Gets the directory from the Windows Registry"""
     reg_path = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\Steam App 2108330'
     try:
         registry_key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, reg_path, 0,
@@ -46,9 +49,9 @@ def F1_23_DIRECTORY() -> any:
         return None
 
 
-video_path = os.path.join(F1_23_DIRECTORY(), "videos")
+video_path = os.path.join(f1_23_directory(), "videos")
 
 skippable = [
     os.path.join(video_path, "attract.bk2"),
     os.path.join(video_path, "cm_f1_sting.bk2")
-]
+]