Rito game client live data integration (#111)

* Add more League API integrations to reliably check death of the user

* Write root certificate to file

* git-ignore certificate

* Repeatedly check for exit buttons when dead

* Consider method of running when creating files

* Remove legacy death image check

* Fix reading the certificate path

.gitignore

@@ -134,3 +134,6 @@ dmypy.json
 screenshots/
 *.log
 bot_settings.ini
+
+# Certificates
+*.pem

constants.py

@@ -56,7 +56,6 @@
             },
         },
         "continue": "captures/buttons/continue.png",
-        "death": "captures/death.png",
         "reconnect": "captures/buttons/reconnect.png",
         "key_fragment": {
             "one": "captures/buttons/key_fragment.png",

lcu_integration.py → league_api_integration.py

@@ -1,21 +1,55 @@
 """
-Integrates the bot with League Client Update (LCU) API.
-Note that the API is allowed to use, but not officially supported by Rito.
-The endpoints we use are stable-ish, and we do not expect them to change soonTM.
+Integrations with the Rito API to have the most reliable data where possible.
 """
 import time
 
 from loguru import logger
 from psutil import Process
 from psutil import process_iter
 import requests
+from requests import HTTPError
 
 # Potentially make this configurable in the future
 # to let the user select their preferred tft mode.
-from requests import HTTPError
-import urllib3
-
 TFT_NORMAL_GAME_QUEUE_ID = 1090
+ROOT_CERTIFICATE = """-----BEGIN CERTIFICATE-----
+MIIEIDCCAwgCCQDJC+QAdVx4UDANBgkqhkiG9w0BAQUFADCB0TELMAkGA1UEBhMC
+VVMxEzARBgNVBAgTCkNhbGlmb3JuaWExFTATBgNVBAcTDFNhbnRhIE1vbmljYTET
+MBEGA1UEChMKUmlvdCBHYW1lczEdMBsGA1UECxMUTG9MIEdhbWUgRW5naW5lZXJp
+bmcxMzAxBgNVBAMTKkxvTCBHYW1lIEVuZ2luZWVyaW5nIENlcnRpZmljYXRlIEF1
+dGhvcml0eTEtMCsGCSqGSIb3DQEJARYeZ2FtZXRlY2hub2xvZ2llc0ByaW90Z2Ft
+ZXMuY29tMB4XDTEzMTIwNDAwNDgzOVoXDTQzMTEyNzAwNDgzOVowgdExCzAJBgNV
+BAYTAlVTMRMwEQYDVQQIEwpDYWxpZm9ybmlhMRUwEwYDVQQHEwxTYW50YSBNb25p
+Y2ExEzARBgNVBAoTClJpb3QgR2FtZXMxHTAbBgNVBAsTFExvTCBHYW1lIEVuZ2lu
+ZWVyaW5nMTMwMQYDVQQDEypMb0wgR2FtZSBFbmdpbmVlcmluZyBDZXJ0aWZpY2F0
+ZSBBdXRob3JpdHkxLTArBgkqhkiG9w0BCQEWHmdhbWV0ZWNobm9sb2dpZXNAcmlv
+dGdhbWVzLmNvbTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKoJemF/
+6PNG3GRJGbjzImTdOo1OJRDI7noRwJgDqkaJFkwv0X8aPUGbZSUzUO23cQcCgpYj
+21ygzKu5dtCN2EcQVVpNtyPuM2V4eEGr1woodzALtufL3Nlyh6g5jKKuDIfeUBHv
+JNyQf2h3Uha16lnrXmz9o9wsX/jf+jUAljBJqsMeACOpXfuZy+YKUCxSPOZaYTLC
+y+0GQfiT431pJHBQlrXAUwzOmaJPQ7M6mLfsnpHibSkxUfMfHROaYCZ/sbWKl3lr
+ZA9DbwaKKfS1Iw0ucAeDudyuqb4JntGU/W0aboKA0c3YB02mxAM4oDnqseuKV/CX
+8SQAiaXnYotuNXMCAwEAATANBgkqhkiG9w0BAQUFAAOCAQEAf3KPmddqEqqC8iLs
+lcd0euC4F5+USp9YsrZ3WuOzHqVxTtX3hR1scdlDXNvrsebQZUqwGdZGMS16ln3k
+WObw7BbhU89tDNCN7Lt/IjT4MGRYRE+TmRc5EeIXxHkQ78bQqbmAI3GsW+7kJsoO
+q3DdeE+M+BUJrhWorsAQCgUyZO166SAtKXKLIcxa+ddC49NvMQPJyzm3V+2b1roP
+SvD2WV8gRYUnGmy/N0+u6ANq5EsbhZ548zZc+BI4upsWChTLyxt2RxR7+uGlS1+5
+EcGfKZ+g024k/J32XP4hdho7WYAS2xMiV83CfLR/MNi8oSMaVQTdKD8cpgiWJk3L
+XWehWA==
+-----END CERTIFICATE-----
+"""
+ROOT_CERTIFICATE_PATH = ""
+
+
+def write_root_certificate_to_file(path: str) -> None:
+    """
+    Writes the Riot root certificate to a file in the file system, because requests needs it to be a file.
+    We need the certificate for SSL connections to any of Riot's APIs.
+    """
+    with open(file=f"{path}\\riotgames_root_certificate.pem", mode="w", encoding="UTF-8") as file:
+        file.write(ROOT_CERTIFICATE)
+    global ROOT_CERTIFICATE_PATH
+    ROOT_CERTIFICATE_PATH = path
 
 
 # LCU logic taken from https://github.com/elliejs/Willump
@@ -58,7 +92,9 @@ def _get_lcu_commandline_arguments(lcu_process: Process):
 
 class LCUIntegration:
     """
-    LCU integration as a class enables us to properly cache and access the session and other variables we might re-use.
+    Integrates the bot with League Client Update (LCU) API.
+    Note that the API is allowed to use, but not officially supported by Rito.
+    The endpoints we use are stable-ish, and we do not expect them to change soonTM.
     """
 
     def __init__(self):
@@ -109,9 +145,7 @@ def connect_to_lcu(self, wait_for_availability: bool = False) -> bool:
                 "Accept": "application/json",
             }
         )
-        # TODO Do proper SSL integration  # pylint: disable=fixme
-        self._session.verify = False
-        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+        self._session.verify = f"{ROOT_CERTIFICATE_PATH}\\riotgames_root_certificate.pem"
 
         logger.info("League client found, trying to connect to it (~60s timeout)")
         timeout = 0
@@ -313,3 +347,51 @@ def should_reconnect(self) -> bool:
             return False
 
         return session_response.json()["phase"] == "Reconnect"
+
+
+class GameClientIntegration:
+    """
+    Class to integrate with the official Rito Game Client API.
+    Sadly the TFT endpoint is re-using the normal League data format.
+    At the moment the only useful data returned are the player health and the player level.
+    """
+
+    def __init__(self):
+        self._url = "https://127.0.0.1:2999"
+        self._session = None
+
+    def create_live_game_client_session(self) -> None:
+        """
+        Creates and caches a session that sets default headers and holds the Rito root SSL certificate.
+        """
+        logger.debug("Creating a session object that holds the root SSL certificate")
+        self._session = requests.Session()
+        self._session.headers.update(
+            {
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            }
+        )
+        self._session.verify = f"{ROOT_CERTIFICATE_PATH}\\riotgames_root_certificate.pem"
+
+    def is_dead(self) -> bool:
+        """
+        Checks if the user is considered dead, aka. has less than or equal to 0 HP.
+
+        Returns:
+            True if the user has less than or equal to 0 HP, else False
+
+        """
+        logger.debug("Checking if we have more than 0 HP")
+        if not self._session:
+            self.create_live_game_client_session()
+
+        active_player_response = self._session.get(f"{self._url}/liveclientdata/activeplayer")
+        try:
+            active_player_response.raise_for_status()
+        except HTTPError:
+            logger.debug("There was an error in the response, assuming that we are dead")
+            return True
+
+        active_player_data = active_player_response.json()
+        return active_player_data["championStats"]["currentHealth"] <= 0.0

tft.py

@@ -11,6 +11,7 @@
 
 import keyboard
 from loguru import logger
+import psutil
 import pyautogui as auto
 import pydirectinput
 
@@ -24,7 +25,7 @@
 from constants import league_processes
 from constants import message_exit_buttons
 from constants import wanted_traits
-import lcu_integration
+import league_api_integration
 from screen_helpers import onscreen
 from screen_helpers import onscreen_multiple_any
 from screen_helpers import onscreen_region_num_loop
@@ -40,7 +41,8 @@
     "VERBOSE": False,
     "OVERRIDE_INSTALL_DIR": None,
 }
-LCU_INTEGRATION = lcu_integration.LCUIntegration()
+LCU_INTEGRATION = league_api_integration.LCUIntegration()
+GAME_CLIENT_INTEGRATION = league_api_integration.GameClientIntegration()
 
 
 def bring_league_client_to_forefront() -> None:
@@ -401,28 +403,56 @@ def exit_now_conditional() -> bool:
     return not league_game_already_running()
 
 
-def check_if_game_complete() -> bool:
-    """Check if the League game is complete.
+def check_screen_for_exit_button() -> bool:
+    """
+    Checks the screen for any exit buttons we expect to appear when the player dies.
 
     Returns:
-        bool: True if any scenario in which the game is not active, False otherwise.
-    """
-    if onscreen(CONSTANTS["client"]["death"]):
-        logger.info("Death detected")
-        click_to_middle(CONSTANTS["client"]["death"])
-        time.sleep(5)
-        return True
+        True if any known exit buttons were found, False if not.
 
+    """
     if onscreen_multiple_any(exit_now_images):
         logger.info("End of game detected (exit now)")
         exit_now_bool = click_to_middle_multiple(exit_now_images, conditional_func=exit_now_conditional, delay=1.5)
         logger.debug(f"Exit now clicking success: {exit_now_bool}")
         time.sleep(5)
         return True
 
+    return False
+
+
+def check_if_game_complete(wait_for_exit_buttons: bool = False) -> bool:
+    """Check if the League game is complete.
+
+    Returns:
+        bool: True if any scenario in which the game is not active, False otherwise.
+    """
+    if wait_for_exit_buttons:
+        logger.info("Waiting an additional ~25s for any exit buttons")
+        for _ in range(24):
+            if check_screen_for_exit_button():
+                return True
+            time.sleep(1)
+
+    if check_screen_for_exit_button():
+        return True
+
     if LCU_INTEGRATION.in_game():
         if LCU_INTEGRATION.should_reconnect():
             attempt_reconnect_to_existing_game()
+
+        if GAME_CLIENT_INTEGRATION.is_dead():
+            if not wait_for_exit_buttons:
+                logger.info("The game considers us dead, checking the screen again for exit buttons")
+                return check_if_game_complete(wait_for_exit_buttons=True)
+
+            logger.warning("We are in-game and considered dead, but no death button has been found.")
+            logger.info("Clicking approximate 'Exit Now' location, please ensure your game is in English.")
+            auto.moveTo(827, 553)
+            click_left()
+            time.sleep(5)
+            return True
+
         return False
 
     return not check_if_client_error()
@@ -763,10 +793,16 @@ def main():
             level="INFO",
         )
 
+    storage_path = "output"
+    for process in psutil.process_iter():
+        if process.name() in {"TFT Bot", "TFT Bot.exe"}:
+            storage_path = system_helpers.expand_environment_variables(CONSTANTS["storage"]["appdata"])
+            break
+
     # File logging, writes to a file in the same folder as the executable.
     # Logs at level DEBUG, so it's always verbose.
     # retention=10 to only keep the 10 most recent files.
-    logger.add("tft-bot-debug-{time}.log", level="DEBUG", retention=10)
+    logger.add(storage_path + "\\tft-bot-debug-{time}.log", level="DEBUG", retention=10)
 
     system_helpers.disable_quickedit()
     # Start auth + main script
@@ -811,7 +847,9 @@ def main():
 
     setup_hotkeys()
 
-    if not lcu_integration.get_lcu_process():
+    league_api_integration.write_root_certificate_to_file(path=storage_path)
+
+    if not league_api_integration.get_lcu_process():
         logger.warning("League client is not open, attempting to start it")
         league_directory = system_helpers.determine_league_install_location(CONFIG["OVERRIDE_INSTALL_DIR"])
         update_league_constants(league_directory)