diff --git a/lolbot/bot/bot.py b/lolbot/bot/bot.py new file mode 100644 index 0000000..58535aa --- /dev/null +++ b/lolbot/bot/bot.py @@ -0,0 +1,334 @@ +""" +Controls the League Client and continually starts League of Legends games. +""" + +import logging +import multiprocessing as mp +import os +import random +import shutil +import traceback +from datetime import datetime, timedelta +from time import sleep + +import pyautogui + +from lolbot.bot import game, launcher, logger, window +from lolbot.common import accounts, config, proc +from lolbot.lcu.lcu_api import LCUApi, LCUError + +log = logging.getLogger(__name__) + +# Click Ratios +POST_GAME_OK_RATIO = (0.4996, 0.9397) +POST_GAME_SELECT_CHAMP_RATIO = (0.4977, 0.5333) +POPUP_SEND_EMAIL_X_RATIO = (0.6960, 0.1238) + +# Errors +MAX_BOT_ERRORS = 5 +MAX_PHASE_ERRORS = 20 + + +class BotError(Exception): + """Indicates the League Client instance should be restarted.""" + pass + + +class Bot: + """Handles the League Client and all tasks needed to start a new game.""" + def __init__(self) -> None: + self.api = LCUApi() + self.config = config.load_config() + self.league_dir = self.config['league_dir'] + self.max_level = self.config['max_level'] + self.lobby = self.config['lobby'] + self.champs = self.config['champs'] + self.dialog = self.config['dialog'] + self.account = None + self.phase = None + self.prev_phase = None + self.bot_errors = 0 + self.phase_errors = 0 + self.game_errors = 0 + + def run(self, message_queue: mp.Queue, games: mp.Value, errors: mp.Value) -> None: + """Main loop, gets an account, launches league, monitors account level, and repeats.""" + logger.MultiProcessLogHandler(message_queue).set_logs() + self.api.update_auth_timer() + self.print_ascii() + # self.wait_for_patching() + self.set_game_config() + while True: + try: + errors.value = self.bot_errors + self.account = accounts.get_account(self.max_level) + launcher.launch_league(self.account['username'], self.account['password']) + self.leveling_loop(games) + proc.close_all_processes() + self.bot_errors = 0 + self.phase_errors = 0 + self.game_errors = 0 + except BotError as be: + log.error(be) + self.bot_errors += 1 + self.phase_errors = 0 + self.game_errors = 0 + if self.bot_errors == MAX_BOT_ERRORS: + log.error("Max errors reached. Exiting") + return + else: + proc.close_all_processes() + except launcher.LaunchError as le: + log.error(le) + log.error("Launcher Error. Exiting") + return + except Exception as e: + log.error(e) + if traceback.format_exc() is not None: + log.error(traceback.format_exc()) + log.error("Unknown Error. Exiting") + return + + def leveling_loop(self, games: mp.Value) -> None: + """Loop that takes action based on the phase of the League Client, continuously starts games.""" + while not self.account_leveled(): + match self.get_phase(): + case 'None' | 'Lobby': + self.start_matchmaking() + case 'Matchmaking': + self.queue() + case 'ReadyCheck': + self.accept_match() + case 'ChampSelect': + self.champ_select() + case 'InProgress': + game.play_game() + case 'Reconnect': + self.reconnect() + case 'WaitingForStats': + self.wait_for_stats() + case 'PreEndOfGame': + self.pre_end_of_game() + case 'EndOfGame': + self.end_of_game() + games.value += 1 + case _: + raise BotError("Unknown phase. {}".format(self.phase)) + + def get_phase(self) -> str: + """Requests the League Client phase.""" + err = None + for i in range(15): + try: + self.prev_phase = self.phase + self.phase = self.api.get_phase() + if self.prev_phase == self.phase and self.phase != "Matchmaking" and self.phase != 'ReadyCheck': + self.phase_errors += 1 + if self.phase_errors == MAX_PHASE_ERRORS: + raise BotError("Transition error. Phase will not change") + else: + self.phase_errors = 0 + sleep(1.5) + return self.phase + except LCUError as e: + err = e + raise BotError(f"Could not get phase: {err}") + + def start_matchmaking(self) -> None: + """Starts matchmaking for a particular game mode, will also wait out dodge timers.""" + # Create lobby + lobby_name = "" + for lobby, lid in config.ALL_LOBBIES.items(): + if lid == self.lobby: + lobby_name = lobby + " " + log.info(f"Creating {lobby_name.lower()}lobby") + try: + self.api.create_lobby(self.lobby) + sleep(3) + except LCUError: + return + + # Start Matchmaking + log.info("Starting queue") + try: + self.api.start_matchmaking() + sleep(1) + except LCUError: + return + + # Wait out dodge timer + try: + time_remaining = self.api.get_dodge_timer() + if time_remaining > 0: + log.info(f"Dodge Timer. Time Remaining: {time_remaining}") + sleep(time_remaining) + except LCUError: + return + + # TODO when queue times get too high switch to pvp lobby, start it, and then switch back + + def queue(self) -> None: + """Waits until the League Client Phase changes to something other than 'Matchmaking'.""" + log.info("Waiting for Ready Check") + start = datetime.now() + while True: + try: + if self.api.get_phase() != 'Matchmaking': + return + elif datetime.now() - start > timedelta(minutes=15): + raise BotError("Queue Timeout") + sleep(1) + except LCUError: + sleep(1) + + def accept_match(self) -> None: + """Accepts the Ready Check.""" + try: + if self.prev_phase != "ReadyCheck": + log.info("Accepting Match") + self.api.accept_match() + except LCUError: + pass + + def champ_select(self) -> None: + """Handles the Champ Select Lobby.""" + log.info("Locking in champ") + champ_index = -1 + while True: + try: + data = self.api.get_champ_select_data() + champ_list = self.champs + self.api.get_available_champion_ids() + except LCUError: + return + try: + for action in data['actions'][0]: + if action['actorCellId'] == data['localPlayerCellId']: + if action['championId'] == 0: # No champ hovered. Hover a champion. + champ_index += 1 + self.api.hover_champion(action['id'], champ_list[champ_index]) + elif not action['completed']: # Champ is hovered but not locked in. + self.api.lock_in_champion(action['id'], action['championId']) + else: # Champ is locked in. Nothing left to do. + sleep(2) + except LCUError: + pass + + def reconnect(self) -> None: + """Attempts to reconnect to an ongoing League of Legends match.""" + log.info("Reconnecting to game") + for i in range(3): + try: + self.api.game_reconnect() + sleep(3) + return + except LCUError: + sleep(2) + log.warning('Could not reconnect to game') + + def wait_for_stats(self) -> None: + """Waits for the League Client Phase to change to something other than 'WaitingForStats'.""" + log.info("Waiting for stats") + for i in range(60): + sleep(2) + try: + if self.api.get_phase() != 'WaitingForStats': + return + except LCUError: + pass + raise BotError("Waiting for stats timeout") + + def pre_end_of_game(self) -> None: + """Handles league of legends client reopening after a game, honoring teammates, and clearing level-up/mission rewards.""" + log.info("Honoring teammates and accepting rewards") + sleep(3) + try: + proc.click(POPUP_SEND_EMAIL_X_RATIO, proc.LEAGUE_CLIENT_WINNAME, 2) + if not self.honor_player(): + sleep(60) # Honor failed for some reason, wait out the honor screen + proc.click(POPUP_SEND_EMAIL_X_RATIO, proc.LEAGUE_CLIENT_WINNAME, 2) + for i in range(3): + proc.click(POST_GAME_SELECT_CHAMP_RATIO, proc.LEAGUE_CLIENT_WINNAME, 1) + proc.click(POST_GAME_OK_RATIO, proc.LEAGUE_CLIENT_WINNAME, 1) + proc.click(POPUP_SEND_EMAIL_X_RATIO, proc.LEAGUE_CLIENT_WINNAME, 1) + except (window.WindowNotFound, pyautogui.FailSafeException): + sleep(3) + + def honor_player(self) -> bool: + """Honors a player in the post game lobby. There are no enemies to honor in bot lobbies.""" + for i in range(3): + try: + players = self.api.get_players_to_honor() + index = random.randint(0, len(players) - 1) + self.api.honor_player(players[index]['summonerId']) + sleep(2) + return True + except LCUError as e: + log.warning(e) + log.warning('Honor Failure') + return False + + def end_of_game(self) -> None: + """Transitions out of EndOfGame.""" + log.info("Getting back into queue") + posted = False + for i in range(15): + try: + if self.api.get_phase() != 'EndOfGame': + return + if not posted: + self.api.play_again() + else: + self.start_matchmaking() + posted = not posted + sleep(1) + except LCUError: + pass + raise BotError("Could not exit play-again screen") + + def account_leveled(self) -> bool: + """Checks if account has reached max level.""" + try: + if self.api.get_summoner_level() >= self.max_level: + if self.account['username'] == self.api.get_display_name(): + self.account['level'] = self.max_level + accounts.save_or_add(self.account) + log.info("Account successfully leveled") + return True + return False + except LCUError: + return False + + def wait_for_patching(self) -> None: + """Checks if the League Client is patching and waits till it is finished.""" + log.info("Checking for Client Updates") + logged = False + while self.api.is_client_patching(): + if not logged: + log.info("Client is patching...") + logged = True + sleep(3) + log.info("Client is up to date") + + def set_game_config(self) -> None: + """Overwrites the League of Legends game config.""" + log.info("Overwriting game configs") + path = self.league_dir + '/Config/game.cfg' + folder = os.path.abspath(os.path.join(path, os.pardir)) + for filename in os.listdir(folder): + file_path = os.path.join(folder, filename) + try: + if os.path.isfile(file_path) or os.path.islink(file_path): + os.unlink(file_path) + except Exception as e: + log.error('Failed to delete %s. Reason: %s' % (file_path, e)) + shutil.copy(proc.resource_path(config.GAME_CFG), path) + + @staticmethod + def print_ascii() -> None: + """Prints some League ascii art.""" + print("""\n\n + ──────▄▌▐▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▌ + ───▄▄██▌█ BEEP BEEP + ▄▄▄▌▐██▌█ -15 LP DELIVERY + ███████▌█▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▌ + ▀(⊙)▀▀▀▀▀▀▀(⊙)(⊙)▀▀▀▀▀▀▀▀▀▀(⊙)\n\n\t\t\t\tLoL Bot\n\n""") diff --git a/lolbot/bot/client.py b/lolbot/bot/client.py deleted file mode 100644 index 8827039..0000000 --- a/lolbot/bot/client.py +++ /dev/null @@ -1,389 +0,0 @@ -""" -Controls the League Client and continually starts League of Legends games -""" - -import os -import shutil -import logging -import random -import traceback -import inspect -from time import sleep -from datetime import datetime, timedelta - -import pyautogui - -import lolbot.bot.launcher as launcher -from lolbot.common import api, utils -from lolbot.bot.game import Game -from lolbot.common.account import AccountManager -from lolbot.common.config import Constants, ConfigRW -from lolbot.common.handler import MultiProcessLogHandler - - -class ClientError(Exception): - """Indicates the League Client instance should be restarted""" - def __init__(self, msg: str = ''): - self.msg = msg - - def __str__(self): - return self.msg - - -class Client: - """Client class that handles the League Client and all tasks needed to start a new game""" - - POST_GAME_OK_RATIO = (0.4996, 0.9397) - POST_GAME_SELECT_CHAMP_RATIO = (0.4977, 0.5333) - POPUP_SEND_EMAIL_X_RATIO = (0.6960, 0.1238) - MAX_CLIENT_ERRORS = 5 - MAX_PHASE_ERRORS = 20 - - def __init__(self, message_queue) -> None: - self.handler = MultiProcessLogHandler(message_queue, Constants.LOG_DIR) - self.log = logging.getLogger(__name__) - self.handler.set_logs() - self.account_manager = AccountManager() - self.connection = api.Connection() - self.launcher = launcher.Launcher() - self.config = ConfigRW() - self.max_level = self.config.get_data('max_level') - self.lobby = self.config.get_data('lobby') - self.champs = self.config.get_data('champs') - self.dialog = self.config.get_data('dialog') - self.account = None - self.phase = "" - self.prev_phase = None - self.client_errors = 0 - self.phase_errors = 0 - self.game_errors = 0 - utils.print_ascii() - self.account_loop() - - def account_loop(self) -> None: - """Main loop, gets an account, launches league, levels the account, and repeats""" - while True: - try: - self.account = self.account_manager.get_account(self.max_level) - self.launcher.launch_league(self.account.username, self.account.password) - self.leveling_loop() - if self.launcher.verify_account(): - self.account_manager.set_account_as_leveled(self.account, self.max_level) - utils.close_all_processes() - self.client_errors = 0 - self.phase_errors = 0 - self.game_errors = 0 - except ClientError as ce: - self.log.error(ce.__str__()) - self.client_errors += 1 - self.phase_errors = 0 - self.game_errors = 0 - if self.client_errors == Client.MAX_CLIENT_ERRORS: - err_msg = "Max errors reached. Exiting" - self.log.error(err_msg) - raise ClientError(err_msg) - utils.close_all_processes() - except launcher.LauncherError as le: - self.log.error(le.__str__()) - self.log.error("Launcher Error. Exiting") - return - except Exception as e: - self.log.error(e) - if traceback.format_exc() is not None: - self.log.error(traceback.format_exc()) - self.log.error("Unknown Error. Exiting") - return - - def leveling_loop(self) -> None: - """Loop that runs the correct function based on the phase of the League Client, continuously starts games""" - self.connection.connect_lcu(verbose=False) - phase = self.get_phase() - if phase != 'InProgress' and phase != 'Reconnect': - self.check_patch() - self.set_game_config() - while not self.account_leveled(): - match self.get_phase(): - case 'None': - self.create_lobby(self.lobby) - case 'Lobby': - self.start_matchmaking(self.lobby) - case 'Matchmaking': - self.queue() - case 'ReadyCheck': - self.accept_match() - case 'ChampSelect': - self.game_lobby() - case 'InProgress': - game: Game = Game() - if self.game_errors == 5: - raise ClientError("Game issue. Most likely client disconnect..") - if not game.play_game(): - self.game_errors += 1 - else: - self.game_errors = 0 - case 'Reconnect': - self.reconnect() - case 'WaitingForStats': - self.wait_for_stats() - case 'PreEndOfGame': - self.pre_end_of_game() - case 'EndOfGame': - self.end_of_game() - case _: - raise ClientError("Unknown phase. {}".format(self.phase)) - - def get_phase(self) -> str: - """Requests the League Client phase""" - for i in range(15): - r = self.connection.request('get', '/lol-gameflow/v1/gameflow-phase') - if r.status_code == 200: - self.prev_phase = self.phase - self.phase = r.json() - self.log.debug("New Phase: {}, Previous Phase: {}".format(self.phase, self.prev_phase)) - if self.prev_phase == self.phase and self.phase != "Matchmaking": - self.phase_errors += 1 - if self.phase_errors == Client.MAX_PHASE_ERRORS: - raise ClientError("Transition error. Phase will not change") - else: - self.log.debug("Phase same as previous. Phase: {}, Previous Phase: {}, Errno {}".format(self.phase, self.prev_phase, self.phase_errors)) - else: - self.phase_errors = 0 - sleep(1.5) - return self.phase - sleep(1) - raise ClientError("Could not get phase") - - def create_lobby(self, lobby_id: int) -> None: - """Creates a lobby for given lobby ID""" - self.log.info("Creating lobby with lobby_id: {}".format(lobby_id)) - self.connection.request('post', '/lol-lobby/v2/lobby', data={'queueId': lobby_id}) - sleep(1.5) - - def start_matchmaking(self, lobby_id: int) -> None: - """Starts matchmaking for a given lobby ID, will also wait out dodge timers""" - self.log.info("Starting queue for lobby_id: {}".format(lobby_id)) - r = self.connection.request('get', '/lol-lobby/v2/lobby') - if r.json()['gameConfig']['queueId'] != lobby_id: - self.create_lobby(lobby_id) - sleep(1) - self.connection.request('post', '/lol-lobby/v2/lobby/matchmaking/search') - sleep(1.5) - - # Check for dodge timer - r = self.connection.request('get', '/lol-matchmaking/v1/search') - if r.status_code == 200 and len(r.json()['errors']) != 0: - dodge_timer = int(r.json()['errors'][0]['penaltyTimeRemaining']) - self.log.info("Dodge Timer. Time Remaining: {}".format(utils.seconds_to_min_sec(dodge_timer))) - sleep(dodge_timer) - - if r.status_code == 200: - if float(r.json()['estimatedQueueTime']) > 6000: - self.log.warning("Queue times are too long") - - def queue(self) -> None: - """Waits until the League Client Phase changes to something other than 'Matchmaking'""" - self.log.info("In queue. Waiting for match") - start = datetime.now() - while True: - if self.get_phase() != 'Matchmaking': - return - elif datetime.now() - start > timedelta(minutes=15): - raise ClientError("Queue Timeout") - elif datetime.now() - start > timedelta(minutes=10): - self.connection.request('delete', '/lol-lobby/v2/lobby/matchmaking/search') - sleep(1) - - def accept_match(self) -> None: - """Accepts the Ready Check""" - self.log.info("Accepting match") - self.connection.request('post', '/lol-matchmaking/v1/ready-check/accept') - - def game_lobby(self) -> None: - """Handles the Champ Select Lobby""" - self.log.info("Lobby opened, picking champ") - r = self.connection.request('get', '/lol-champ-select/v1/session') - if r.status_code != 200: - return - cs = r.json() - - r2 = self.connection.request('get', '/lol-lobby-team-builder/champ-select/v1/pickable-champion-ids') - if r2.status_code != 200: - return - f2p = r2.json() - - champ_index = 0 - f2p_index = 0 - requested = False - while r.status_code == 200: - lobby_state = cs['timer']['phase'] - lobby_time_left = int(float(cs['timer']['adjustedTimeLeftInPhase']) / 1000) - - # Find player action - for action in cs['actions'][0]: # There are 5 actions in the first action index, one for each player - if action['actorCellId'] != cs['localPlayerCellId']: # determine which action corresponds to bot - continue - - # Check if champ is already locked in - if not action['completed']: - # Select Champ or Lock in champ that has already been selected - if action['championId'] == 0: # no champ selected, attempt to select a champ - self.log.debug("Lobby State: {}. Time Left in Lobby: {}s. Action: Hovering champ".format(lobby_state, lobby_time_left)) - if champ_index < len(self.champs): - champion_id = self.champs[champ_index] - champ_index += 1 - else: - champion_id = f2p[f2p_index] - f2p_index += 1 - url = '/lol-champ-select/v1/session/actions/{}'.format(action['id']) - data = {'championId': champion_id} - self.connection.request('patch', url, data=data) - else: # champ selected, lock in - self.log.debug("Lobby State: {}. Time Left in Lobby: {}s. Action: Locking in champ".format(lobby_state, lobby_time_left)) - url = '/lol-champ-select/v1/session/actions/{}'.format(action['id']) - data = {'championId': action['championId']} - self.connection.request('post', url + '/complete', data=data) - - # Ask for mid - if not requested: - sleep(1) - try: - self.chat(random.choice(self.dialog)) - except IndexError: - pass - requested = True - else: - self.log.debug("Lobby State: {}. Time Left in Lobby: {}s. Action: Waiting".format(lobby_state, lobby_time_left)) - r = self.connection.request('get', '/lol-champ-select/v1/session') - if r.status_code != 200: - self.log.info('Lobby closed') - return - cs = r.json() - sleep(3) - - def reconnect(self) -> None: - """Attempts to reconnect to an ongoing League of Legends match""" - self.log.info("Reconnecting to game") - for i in range(3): - r = self.connection.request('post', '/lol-gameflow/v1/reconnect') - if r.status_code == 204: - return - sleep(2) - self.log.warning('Could not reconnect to game') - - def wait_for_stats(self) -> None: - """Waits for the League Client Phase to change to something other than 'WaitingForStats'""" - self.log.info("Waiting for stats") - for i in range(60): - sleep(2) - if self.get_phase() != 'WaitingForStats': - return - raise ClientError("Waiting for stats timeout") - - def pre_end_of_game(self) -> None: - """Handles league of legends client reopening after a game, honoring teammates, and clearing level-up/mission rewards""" - self.log.info("Honoring teammates and accepting rewards") - sleep(3) - try: - utils.click(Client.POPUP_SEND_EMAIL_X_RATIO, utils.LEAGUE_CLIENT_WINNAME, 2) - self.honor_player() - utils.click(Client.POPUP_SEND_EMAIL_X_RATIO, utils.LEAGUE_CLIENT_WINNAME, 2) - for i in range(3): - utils.click(Client.POST_GAME_SELECT_CHAMP_RATIO, utils.LEAGUE_CLIENT_WINNAME, 1) - utils.click(Client.POST_GAME_OK_RATIO, utils.LEAGUE_CLIENT_WINNAME, 1) - utils.click(Client.POPUP_SEND_EMAIL_X_RATIO, utils.LEAGUE_CLIENT_WINNAME, 1) - except (utils.WindowNotFound, pyautogui.FailSafeException): - sleep(3) - - def end_of_game(self) -> None: - """Transitions League Client to 'Lobby' phase.""" - self.log.info("Post game. Starting a new loop") - posted = False - for i in range(15): - if self.get_phase() != 'EndOfGame': - return - if not posted: - self.connection.request('post', '/lol-lobby/v2/play-again') - else: - self.create_lobby(self.lobby) - posted = not posted - sleep(1) - raise ClientError("Could not exit play-again screen") - - def account_leveled(self) -> bool: - """Checks if account has reached the constants.MAX_LEVEL (default 30)""" - r = self.connection.request('get', '/lol-chat/v1/me') - if r.status_code == 200: - self.account.level = int(r.json()['lol']['level']) - if self.account.level < self.max_level: - self.log.debug("Account Level: {}.".format(self.account.level)) - return False - else: - self.log.info("SUCCESS: Account Leveled") - return True - - def check_patch(self) -> None: - """Checks if the League Client is patching and waits till it is finished""" - self.log.info("Checking for Client Updates") - r = self.connection.request('get', '/patcher/v1/products/league_of_legends/state') - if r.status_code != 200: - return - logged = False - while not r.json()['isUpToDate']: - if not logged: - self.log.info("Client is patching...") - logged = True - sleep(3) - r = self.connection.request('get', '/patcher/v1/products/league_of_legends/state') - self.log.debug('Status Code: {}, Percent Patched: {}%'.format(r.status_code, r.json()['percentPatched'])) - self.log.debug(r.json()) - self.log.info("Client is up to date") - - def honor_player(self) -> None: - """Honors a player in the post game lobby""" - for i in range(3): - r = self.connection.request('get', '/lol-honor-v2/v1/ballot') - if r.status_code == 200: - players = r.json()['eligibleAllies'] - index = random.randint(0, len(players)-1) - self.connection.request('post', '/lol-honor-v2/v1/honor-player', data={"summonerId": players[index]['summonerId']}) - self.log.debug("Honor Success: Player {}. Champ: {}. Summoner: {}. ID: {}".format(index+1, players[index]['championName'], players[index]['summonerName'], players[index]['summonerId'])) - sleep(2) - return - sleep(2) - self.log.warning('Honor Failure. Player -1, Champ: NULL. Summoner: NULL. ID: -1') - self.connection.request('post', '/lol-honor-v2/v1/honor-player', data={"summonerId": 0}) # will clear honor screen - - def chat(self, msg: str) -> None: - """Sends a message to the chat window""" - chat_id = '' - r = self.connection.request('get', '/lol-chat/v1/conversations') - if r.status_code != 200: - self.log.warning("{} chat attempt failed. Could not reach endpoint".format(inspect.stack()[1][3])) - return - for convo in r.json(): - if convo['gameName'] != '' and convo['gameTag'] != '': - continue - chat_id = convo['id'] - if chat_id == '': - self.log.warning('{} chat attempt failed. Could not send message. Chat ID is Null'.format(inspect.stack()[1][3])) - return - data = {"body": msg} - r = self.connection.request('post', '/lol-chat/v1/conversations/{}/messages'.format(chat_id), data=data) - if r.status_code != 200: - self.log.warning('Could not send message. HTTP STATUS: {} - {}, Caller: {}'.format(r.status_code, r.json(), inspect.stack()[1][3])) - else: - self.log.debug("Message success. Msg: {}. Caller: {}".format(msg, inspect.stack()[1][3])) - - def set_game_config(self) -> None: - """Overwrites the League of Legends game config""" - self.log.info("Overwriting game configs") - path = self.config.get_data('league_config') - folder = os.path.abspath(os.path.join(path, os.pardir)) - for filename in os.listdir(folder): - file_path = os.path.join(folder, filename) - try: - if os.path.isfile(file_path) or os.path.islink(file_path): - os.unlink(file_path) - except Exception as e: - print('Failed to delete %s. Reason: %s' % (file_path, e)) - shutil.copy(utils.resource_path(Constants.GAME_CFG), path) diff --git a/lolbot/bot/controller.py b/lolbot/bot/controller.py new file mode 100644 index 0000000..c95f500 --- /dev/null +++ b/lolbot/bot/controller.py @@ -0,0 +1,63 @@ +""" +Module that handles clicks and key presses. +""" + +from time import sleep + +import keyboard +import mouse # pyautogui clicks do not work with league/directx +import pyautogui + +from lolbot.bot.window import * + + +def keypress(key: str, window: str = '', wait: float = 1) -> None: + """Sends a keypress to a window""" + if window != '' and not window_exists(window): + raise WindowNotFound + keyboard.press_and_release(key) + sleep(wait) + + +def write(keys: str, window: str = '', wait: float = 1) -> None: + """Sends a string of key presses to a window""" + if window != '' and not window_exists(window): + raise WindowNotFound + pyautogui.write(keys, interval=0.11) + sleep(wait) + + +def left_click(ratio: tuple, window: str, wait: float = 1) -> None: + """Makes a click in an open window""" + _move_to_window_coords(ratio, window) + mouse.click() + sleep(wait) + + +def right_click(ratio: tuple, window: str, wait: float = 1) -> None: + """Makes a right click in an open window""" + _move_to_window_coords(ratio, window) + mouse.right_click() + sleep(wait) + + +def attack_move_click(ratio: tuple, window: str, wait: float = 1) -> None: + """Attack move clicks in an open window""" + _move_to_window_coords(ratio, window) + keyboard.press('a') + sleep(.1) + mouse.click() + sleep(.1) + mouse.click() + keyboard.release('a') + sleep(wait) + + +def _move_to_window_coords(ratio: tuple, window: str): + if not window_exists(window): + raise WindowNotFound + x, y, l, h = get_window_size(window) + updated_x = ((l - x) * ratio[0]) + x + updated_y = ((h - y) * ratio[1]) + y + pyautogui.moveTo(updated_x, updated_y) + sleep(.5) diff --git a/lolbot/bot/game.py b/lolbot/bot/game.py index 9e7d832..366300b 100644 --- a/lolbot/bot/game.py +++ b/lolbot/bot/game.py @@ -1,234 +1,167 @@ """ -Plays and monitors the state of a single League of Legends match +Game logic that plays and through a single League of Legends match. """ import logging -import inspect import random -from enum import Enum from datetime import datetime, timedelta -from time import sleep -import pyautogui -import requests +import lolbot.lcu.game_api as api +from lolbot.bot.controller import * +from lolbot.bot.window import game_window_exists, WindowNotFound, GAME_WINDOW_NAME +from lolbot.common import proc -from lolbot.common import utils +log = logging.getLogger(__name__) +# Game Times +LOADING_SCREEN_TIME = 3 +MINION_CLASH_TIME = 85 +FIRST_TOWER_TIME = 630 +MAX_GAME_TIME = 2400 -class GameState(Enum): - LOADING_SCREEN = 0 # 0 sec -> 3 sec - PRE_MINIONS = 1 # 3 sec -> 90 sec - EARLY_GAME = 2 # 90 sec -> constants.EARLY_GAME_END_TIME - LATE_GAME = 3 # constants.EARLY_GAME_END_TIME -> end of game +# Click coordinates to move/aim +MINI_MAP_UNDER_TURRET = (0.8760, 0.8846) +MINI_MAP_CENTER_MID = (0.8981, 0.8674) +MINI_MAP_ENEMY_NEXUS = (0.9628, 0.7852) +ULT_DIRECTION = (0.7298, 0.2689) +CENTER_OF_SCREEN = (0.5, 0.5) + +# Click coordinates to purchase items +AFK_OK_BUTTON = (0.4981, 0.4647) +SYSTEM_MENU_X_BUTTON = (0.7729, 0.2488) +SHOP_ITEM_BUTTONS = [(0.3216, 0.5036), (0.4084, 0.5096), (0.4943, 0.4928)] +SHOP_PURCHASE_ITEM_BUTTON = (0.7586, 0.6012) + +MAX_ERRORS = 15 class GameError(Exception): """Indicates the game should be terminated""" - def __init__(self, msg=''): - self.msg = msg - - def __str__(self): - return self.msg - - -class Game: - """Game class that handles the tasks needed to play/win a bot game of League of Legends""" - - MINI_MAP_UNDER_TURRET = (0.8760, 0.8846) - MINI_MAP_CENTER_MID = (0.8981, 0.8674) - MINI_MAP_ENEMY_NEXUS = (0.9628, 0.7852) - - ULT_DIRECTION = (0.7298, 0.2689) - CENTER_OF_SCREEN = (0.5, 0.5) - - AFK_OK_BUTTON = (0.4981, 0.4647) - SYSTEM_MENU_X_BUTTON = (0.7729, 0.2488) - SHOP_ITEM_BUTTONS = [(0.3216, 0.5036), (0.4084, 0.5096), (0.4943, 0.4928)] - SHOP_PURCHASE_ITEM_BUTTON = (0.7586, 0.6012) - - EARLY_GAME_END_TIME = 630 - MAX_GAME_TIME = 2400 - - def __init__(self) -> None: - self.log = logging.getLogger(__name__) - self.connection_errors = 0 - self.game_data = None - self.game_time = None - self.formatted_game_time = None - self.game_state = None - self.screen_locked = False - self.in_lane = False - self.is_dead = False - self.ability_upgrades = ['ctrl+r', 'ctrl+q', 'ctrl+w', 'ctrl+e'] - - def play_game(self) -> bool: - """Plays a single game of League of Legends, takes actions based on game time""" - try: - self.wait_for_game_window() - self.wait_for_connection() - while True: - self.update_state() - match self.game_state: - case GameState.LOADING_SCREEN: - self.loading_screen() - case GameState.PRE_MINIONS: - self.game_start() - case GameState.EARLY_GAME: - self.play(Game.MINI_MAP_CENTER_MID, Game.MINI_MAP_UNDER_TURRET, 20) - case GameState.LATE_GAME: - self.play(Game.MINI_MAP_ENEMY_NEXUS, Game.MINI_MAP_CENTER_MID, 35) - except GameError as e: - self.log.warning(e.__str__()) - utils.close_game() + pass + + +def play_game() -> None: + """Plays a single game of League of Legends, takes actions based on game time""" + game_errors = 0 + logged = False + try: + wait_for_game_window() + wait_for_connection() + while True: + if api.is_dead(): + sleep(2) + continue + game_time = api.get_game_time() + if game_time > MINION_CLASH_TIME and not logged: + log.info("Destroying Enemy Nexus") + logged = True + if game_time < LOADING_SCREEN_TIME: + loading_screen() + elif game_time < MINION_CLASH_TIME: + game_start() + elif game_time < FIRST_TOWER_TIME: + play(MINI_MAP_CENTER_MID, MINI_MAP_UNDER_TURRET, 20) + elif game_time < MAX_GAME_TIME: + play(MINI_MAP_ENEMY_NEXUS, MINI_MAP_CENTER_MID, 35) + else: + raise GameError("Game has exceeded the max time limit") + except GameError as e: + log.warning(str(e)) + proc.close_game() + sleep(30) + except api.GameAPIError as e: + game_errors += 1 + if game_errors == MAX_ERRORS: + log.error(f"Max Game Errors reached. {e}") + proc.close_game() sleep(30) - return False - except (utils.WindowNotFound, pyautogui.FailSafeException): - self.log.info("Game Complete. Game Time: {}".format(self.formatted_game_time)) - return True - - def wait_for_game_window(self) -> None: - """Loop that waits for game window to open""" - self.log.debug("Waiting for game window to open") - for i in range(120): - sleep(1) - if utils.exists(utils.LEAGUE_GAME_CLIENT_WINNAME): - self.log.debug("Game window open") - utils.click(Game.CENTER_OF_SCREEN, utils.LEAGUE_GAME_CLIENT_WINNAME, 2) - utils.click(Game.CENTER_OF_SCREEN, utils.LEAGUE_GAME_CLIENT_WINNAME) - return - raise GameError("Game window did not open") - - def wait_for_connection(self) -> None: - """Loop that waits for connection to local game server""" - self.log.debug("Connecting to game server...") - for i in range(120): - try: - response = requests.get('https://127.0.0.1:2999/liveclientdata/allgamedata', timeout=10, verify=False) - if response.status_code == 200: - self.log.debug("Connected to game server") - return - except ConnectionError: - pass - sleep(1) - raise GameError("Game window opened but connection failed") - - def loading_screen(self) -> None: - """Loop that waits for loading screen to end""" - self.log.info("In loading screen. Waiting for game to start") - start = datetime.now() - while self.game_time < 3: - if datetime.now() - start > timedelta(minutes=10): - raise GameError("Loading Screen max time limit exceeded") - self.update_state(postpone_update=2) - utils.click(Game.CENTER_OF_SCREEN, utils.LEAGUE_GAME_CLIENT_WINNAME, 2) - - def game_start(self) -> None: - """Buys starter items and waits for minions to clash (minions clash at 90 seconds)""" - self.log.info("Game start. Waiting for minions") - sleep(10) - self.buy_item() - self.lock_screen() - self.upgrade_abilities() - while self.game_state == GameState.PRE_MINIONS: - utils.right_click(Game.MINI_MAP_UNDER_TURRET, utils.LEAGUE_GAME_CLIENT_WINNAME, 2) # to prevent afk warning popup - utils.click(Game.AFK_OK_BUTTON, utils.LEAGUE_GAME_CLIENT_WINNAME) - self.update_state() - self.in_lane = True - - def play(self, attack_position: tuple, retreat_position: tuple, time_to_lane: int) -> None: - """A set of player actions. Buys items, levels up abilities, heads to lane, attacks, then retreats""" - self.log.debug("Main player loop. GameState: {}".format(self.game_state)) - self.buy_item() - self.lock_screen() - self.upgrade_abilities() - while self.is_dead: - self.update_state() - utils.click(Game.AFK_OK_BUTTON, utils.LEAGUE_GAME_CLIENT_WINNAME) - if not self.in_lane: - utils.attack_move_click(attack_position) - utils.press('d', utils.LEAGUE_GAME_CLIENT_WINNAME) # ghost - sleep(time_to_lane) - self.in_lane = True - - # Main attack move loop. This sequence attacks and then de-aggros to prevent them from dying 50 times. - for i in range(7): - utils.attack_move_click(attack_position, 8) - utils.right_click(retreat_position, utils.LEAGUE_GAME_CLIENT_WINNAME, 2.5) - - # Ult and back - utils.press('f', utils.LEAGUE_GAME_CLIENT_WINNAME) - utils.attack_move_click(Game.ULT_DIRECTION) - utils.press('r', utils.LEAGUE_GAME_CLIENT_WINNAME, 4) - utils.right_click(Game.MINI_MAP_UNDER_TURRET, utils.LEAGUE_GAME_CLIENT_WINNAME, 6) - utils.press('b', utils.LEAGUE_GAME_CLIENT_WINNAME, 10) - self.in_lane = False - - def buy_item(self) -> None: - """Opens the shop and attempts to purchase items via default shop hotkeys""" - self.log.debug("Attempting to purchase an item from build order") - utils.press('p', utils.LEAGUE_GAME_CLIENT_WINNAME, 1.5) - utils.click(random.choice(Game.SHOP_ITEM_BUTTONS), utils.LEAGUE_GAME_CLIENT_WINNAME, 1.5) - utils.click(Game.SHOP_PURCHASE_ITEM_BUTTON, utils.LEAGUE_GAME_CLIENT_WINNAME, 1.5) - utils.press('esc', utils.LEAGUE_GAME_CLIENT_WINNAME, 1.5) - utils.click(Game.SYSTEM_MENU_X_BUTTON, utils.LEAGUE_GAME_CLIENT_WINNAME, 1.5) - - def lock_screen(self) -> None: - """Locks screen on champion""" - if not self.screen_locked: - self.log.debug("Locking screen") - utils.press('y', utils.LEAGUE_GAME_CLIENT_WINNAME) - self.screen_locked = True - - def upgrade_abilities(self) -> None: - """Upgrades abilities and then rotates which ability will be upgraded first next time""" - self.log.debug("Upgrading abilities. Second Ability: {}".format(self.ability_upgrades[1])) - for upgrade in self.ability_upgrades: - utils.press(upgrade, utils.LEAGUE_GAME_CLIENT_WINNAME) - self.ability_upgrades = ([self.ability_upgrades[0]] + [self.ability_upgrades[-1]] + self.ability_upgrades[1:-1]) # r is always first - - def update_state(self, postpone_update: int = 1) -> bool: - """Gets game data from local game server and updates game state""" - self.log.debug("Updating state. Caller: {}".format(inspect.stack()[1][3])) - sleep(postpone_update) - try: - response = requests.get('https://127.0.0.1:2999/liveclientdata/allgamedata', timeout=10, verify=False) - except: - self.log.debug("Connection error. Could not get game data") - self.connection_errors += 1 - if not utils.exists(utils.LEAGUE_GAME_CLIENT_WINNAME): - raise utils.WindowNotFound - if self.connection_errors == 15: - raise GameError("Connection Error. Could not connect to game") - return False - if response.status_code != 200: - self.log.debug("Connection error. Response status code: {}".format(response.status_code)) - self.connection_errors += 1 - if not utils.exists(utils.LEAGUE_GAME_CLIENT_WINNAME): - raise utils.WindowNotFound - if self.connection_errors == 15: - raise GameError("Bad Response. Could not connect to game") - return False - - self.game_data = response.json() - for player in self.game_data['allPlayers']: - if player['summonerName'] == self.game_data['activePlayer']['summonerName']: - self.is_dead = bool(player['isDead']) - self.game_time = int(self.game_data['gameData']['gameTime']) - self.formatted_game_time = utils.seconds_to_min_sec(self.game_time) - if self.game_time < 3: - self.game_state = GameState.LOADING_SCREEN - elif self.game_time < 85: - self.game_state = GameState.PRE_MINIONS - elif self.game_time < Game.EARLY_GAME_END_TIME: - if self.game_state != GameState.EARLY_GAME: - self.log.info("Early Game. Pushing center mid. Game Time: {}".format(self.formatted_game_time)) - self.game_state = GameState.EARLY_GAME - elif self.game_time < Game.MAX_GAME_TIME: - if self.game_state != GameState.LATE_GAME: - self.log.info("Mid Game. Pushing enemy nexus. Game Time: {}".format(self.formatted_game_time)) - self.game_state = GameState.LATE_GAME - else: - raise GameError("Game has exceeded the max time limit") - self.connection_errors = 0 - self.log.debug("State Updated. Game Time: {}, Game State: {}, IsDead: {}".format(self.game_time, self.game_state, self.is_dead)) - return True + except (WindowNotFound, pyautogui.FailSafeException): + log.info(f"Game Complete") + + +def wait_for_game_window() -> None: + """Loop that waits for game window to open""" + for i in range(120): + sleep(1) + if game_window_exists(): + log.debug("Game window open") + left_click(CENTER_OF_SCREEN, GAME_WINDOW_NAME, 2) + left_click(CENTER_OF_SCREEN, GAME_WINDOW_NAME) + return + raise GameError("Game window did not open") + + +def wait_for_connection() -> None: + """Loop that waits for connection to local game server""" + for i in range(120): + if api.is_connected(): + return + sleep(1) + raise GameError("Game window opened but connection failed") + + +def loading_screen() -> None: + """Loop that waits for loading screen to end""" + log.info("Waiting for game to start") + start = datetime.now() + while api.get_game_time() < LOADING_SCREEN_TIME: + sleep(2) + if datetime.now() - start > timedelta(minutes=10): + raise GameError("Loading screen max time limit exceeded") + left_click(CENTER_OF_SCREEN, GAME_WINDOW_NAME, 2) + + +def game_start() -> None: + """Buys starter items and waits for minions to clash (minions clash at 90 seconds)""" + log.info("Buying items, heading mid, and waiting for minions") + sleep(10) + shop() + keypress('y', GAME_WINDOW_NAME) # lock screen + upgrade_abilities() + + # Sit under turret till minions clash mid lane + while api.get_game_time() < MINION_CLASH_TIME: + right_click(MINI_MAP_UNDER_TURRET, GAME_WINDOW_NAME, 2) # to prevent afk warning popup + left_click(AFK_OK_BUTTON, GAME_WINDOW_NAME) + + +def play(attack: tuple, retreat: tuple, time_to_lane: int) -> None: + """Buys items, levels up abilities, heads to lane, attacks, retreats, backs""" + shop() + upgrade_abilities() + left_click(AFK_OK_BUTTON, GAME_WINDOW_NAME) + + # Walk to lane + attack_move_click(attack, GAME_WINDOW_NAME) + keypress('d', GAME_WINDOW_NAME) # ghost + sleep(time_to_lane) + + # Main attack move loop. This sequence attacks and then de-aggros to prevent them from dying 50 times. + for i in range(7): + attack_move_click(attack, GAME_WINDOW_NAME, 8) + right_click(retreat, GAME_WINDOW_NAME, 2.5) + + # Ult and back + keypress('f', GAME_WINDOW_NAME) + attack_move_click(ULT_DIRECTION, GAME_WINDOW_NAME) + keypress('r', GAME_WINDOW_NAME, 4) + right_click(MINI_MAP_UNDER_TURRET, GAME_WINDOW_NAME, 6) + keypress('b', GAME_WINDOW_NAME, 10) + + +def shop() -> None: + """Opens the shop and attempts to purchase items via default shop hotkeys""" + keypress('p', GAME_WINDOW_NAME, 1.5) # open shop + left_click(random.choice(SHOP_ITEM_BUTTONS), GAME_WINDOW_NAME, 1.5) + left_click(SHOP_PURCHASE_ITEM_BUTTON, GAME_WINDOW_NAME, 1.5) + keypress('esc', GAME_WINDOW_NAME, 1.5) + left_click(SYSTEM_MENU_X_BUTTON, GAME_WINDOW_NAME, 1.5) + + +def upgrade_abilities() -> None: + """Upgrades abilities and then rotates which ability will be upgraded first next time""" + keypress('ctrl+r', GAME_WINDOW_NAME) + upgrades = ['ctrl+q', 'ctrl+w', 'ctrl+e'] + random.shuffle(upgrades) + for upgrade in upgrades: + keypress(upgrade, GAME_WINDOW_NAME) diff --git a/lolbot/bot/launcher.py b/lolbot/bot/launcher.py index d5a1775..631c332 100644 --- a/lolbot/bot/launcher.py +++ b/lolbot/bot/launcher.py @@ -1,120 +1,110 @@ """ -Handles Riot Client and login to launch the League Client +Handles launching League of Legends and logging into an account. """ import logging import subprocess -from time import sleep from pathlib import Path +from time import sleep -from lolbot.common import api -from lolbot.common import utils -from lolbot.common.config import ConfigRW - - -class LauncherError(Exception): - def __init__(self, msg=''): - self.msg = msg - - def __str__(self): - return self.msg +import lolbot.bot.window as window +from lolbot.bot import controller +from lolbot.common import config, proc +from lolbot.lcu.lcu_api import LCUApi, LCUError -class Launcher: - """Handles the Riot Client and launches League of Legends""" +log = logging.getLogger(__name__) - def __init__(self) -> None: - self.log = logging.getLogger(__name__) - self.connection = api.Connection() - self.config = ConfigRW() - self.username = "" - self.password = "" +MAX_RETRIES = 5 - def launch_league(self, username: str, password: str) -> None: - """Runs setup logic and starts launch sequence""" - if not username or not password: - self.log.warning('No account set. Add accounts on account page') - self.username = username - self.password = password - self.launch_loop() - def launch_loop(self) -> None: - """Handles tasks necessary to open the League of Legends client""" - logged_in = False - for i in range(100): +class LaunchError(Exception): + """Indicates that League could not be opened.""" + pass - # League is running and there was a successful login attempt - if utils.is_league_running() and logged_in: - self.log.info("Launch Success") - utils.close_riot_client() - return - # League is running without a login attempt - elif utils.is_league_running() and not logged_in: - self.log.warning("League opened with prior login") - self.verify_account() +def launch_league(username: str, password: str) -> None: + """Ensures that League is open and logged into a specific account""" + api = LCUApi() + api.update_auth() + login_attempted = False + logins = 0 + for i in range(30): + try: + if proc.is_league_running(): + if login_attempted: + log.info("Launch success") + proc.close_riot_client() + else: + log.warning("League opened with prior login") + verify_account(api, username) return - - # League is not running but Riot Client is running - elif not utils.is_league_running() and utils.is_rc_running(): - # Get session state - self.connection.set_rc_headers() - r = self.connection.request("get", "/rso-auth/v1/authorization/access-token") - - # Already logged in - if r.status_code == 200 and not logged_in: - self.start_league() - - # Not logged in and haven't logged in - if r.status_code == 404 and not logged_in: - self.login() - logged_in = True - sleep(1) - - # Logged in - elif r.status_code == 200 and logged_in: - self.start_league() - - # Nothing is running - elif not utils.is_league_running() and not utils.is_rc_running(): - self.start_league() + elif proc.is_rc_running(): + if api.access_token_exists(): + if not login_attempted: + log.warning("Riot Client already logged in") + try: + api.launch_league_from_rc() + sleep(2) + except LCUError: + pass + continue + else: + if logins == MAX_RETRIES: + raise LaunchError("Max login attempts exceeded. Check username and password") + else: + logins += 1 + log.info("Logging into Riot Client") + login_attempted = True + # api.login(username, password) # they turned this off + manual_login(username, password) + if not api.access_token_exists(): + log.warning("Login attempt failed") + proc.close_riot_client() + sleep(5) + else: + start_league() + sleep(10) + except LCUError: sleep(2) - - if logged_in: - raise LauncherError("Launch Error. Most likely the Riot Client needs an update or League needs an update from within Riot Client") - else: - raise LauncherError("Could not launch League of legends") - - def start_league(self): - self.log.info('Launching League') - rclient = Path(self.config.get_data('league_path')).parent.absolute().parent.absolute() - rclient = str(rclient) + "/Riot Client/RiotClientServices" - subprocess.Popen([rclient, "--launch-product=league_of_legends", "--launch-patchline=live"]) - sleep(3) - - def login(self) -> None: - """Sends account credentials to Riot Client""" - self.log.info("Logging into Riot Client") - body = {"clientId": "riot-client", 'trustLevels': ['always_trusted']} - r = self.connection.request("post", "/rso-auth/v2/authorizations", data=body) - if r.status_code != 200: - raise LauncherError("Failed Authorization Request. Response: {}".format(r.status_code)) - body = {"username": self.username, "password": self.password, "persistLogin": False} - r = self.connection.request("put", '/rso-auth/v1/session/credentials', data=body) - if r.status_code != 201: - raise LauncherError("Failed Authentication Request. Response: {}".format(r.status_code)) - elif r.json()['error'] == 'auth_failure': - raise LauncherError("Invalid username or password") - - def verify_account(self) -> bool: - """Checks if account credentials match the account on the League Client""" - self.log.info("Verifying logged-in account credentials") - connection = api.Connection() - connection.connect_lcu(verbose=False) - r = connection.request('get', '/lol-login/v1/session') - if r.json()['username'] != self.username: - self.log.warning("Accounts do not match! Proceeding anyways") - return False - else: - self.log.info("Account Verified") - return True + raise LaunchError("Could not launch league. Ensure there are no pending updates.") + + +def manual_login(username: str, password: str): + log.info('Manually logging into Riot Client') + window.activate_windw("Riot Client") + controller.write(username) + sleep(.5) + controller.keypress('tab') + sleep(.5) + controller.write(password) + sleep(.5) + controller.keypress('enter') + sleep(10) + + +def start_league(): + """Launches League of Legends from Riot Client.""" + log.info('Launching League of Legends') + c = config.load_config() + riot_client_dir = Path(c['league_dir']).parent.absolute() + riot_client_path = str(riot_client_dir) + "/Riot Client/RiotClientServices" + subprocess.Popen([riot_client_path, "--launch-product=league_of_legends", "--launch-patchline=live"]) + sleep(3) + + +def verify_account(api: LCUApi, username: str = None) -> bool: + """Checks if account username match the account that is currently logged in.""" + log.info("Verifying logged-in account credentials") + name = "" + try: + name = api.get_display_name() + except LCUError: + return False + + if username == name: + log.info("Account Verified") + return True + else: + log.warning("Accounts do not match!") + return False diff --git a/lolbot/common/handler.py b/lolbot/bot/logger.py similarity index 71% rename from lolbot/common/handler.py rename to lolbot/bot/logger.py index c55cfe6..5013709 100644 --- a/lolbot/common/handler.py +++ b/lolbot/bot/logger.py @@ -1,22 +1,22 @@ """ -Handles bot logging +Sets global logging state. """ import logging import os import sys from datetime import datetime +from logging.handlers import RotatingFileHandler from multiprocessing import Queue -from logging.handlers import RotatingFileHandler +import lolbot.common.config as config class MultiProcessLogHandler(logging.Handler): - """Class that handles bot log messsages, writes to log file, terminal, and view""" + """Sets log configuration and pushes logs onto message queue for display in view""" - def __init__(self, message_queue: Queue, path: str) -> None: + def __init__(self, message_queue: Queue) -> None: logging.Handler.__init__(self) - self.log_dir = path self.message_queue = message_queue def emit(self, record: logging.LogRecord) -> None: @@ -26,16 +26,16 @@ def emit(self, record: logging.LogRecord) -> None: def set_logs(self) -> None: """Sets log configurations""" - if not os.path.exists(self.log_dir): - os.makedirs(self.log_dir) + if not os.path.exists(config.LOG_DIR): + os.makedirs(config.LOG_DIR) - filename = os.path.join(self.log_dir, datetime.now().strftime('%d%m%Y_%H%M.log')) + filename = os.path.join(config.LOG_DIR, datetime.now().strftime('%d%m%Y_%H%M.log')) formatter = logging.Formatter(fmt='[%(asctime)s] [%(levelname)-7s] [%(funcName)-21s] %(message)s',datefmt='%d %b %Y %H:%M:%S') - logging.getLogger().setLevel(logging.DEBUG) + logging.getLogger().setLevel(logging.INFO) fh = RotatingFileHandler(filename=filename, maxBytes=1000000, backupCount=1) fh.setFormatter(formatter) - fh.setLevel(logging.DEBUG) + fh.setLevel(logging.INFO) logging.getLogger().addHandler(fh) ch = logging.StreamHandler(sys.stdout) diff --git a/lolbot/bot/window.py b/lolbot/bot/window.py new file mode 100644 index 0000000..72e7db1 --- /dev/null +++ b/lolbot/bot/window.py @@ -0,0 +1,61 @@ +""" +Utility functions for determining if a window exists. +""" + +from win32gui import FindWindow, GetWindowRect +import pygetwindow as gw + +CLIENT_WINDOW_NAME = "League of Legends" +GAME_WINDOW_NAME = "League of Legends (TM) Client" + + +class WindowNotFound(Exception): + pass + + +def game_window_exists() -> bool: + """Checks if the league of legends game window exists""" + if FindWindow(None, GAME_WINDOW_NAME) == 0: + return False + return True + + +def client_window_exists() -> bool: + """Checks if the league of legends client window exists""" + if FindWindow(None, CLIENT_WINDOW_NAME) == 0: + return False + return True + + +def window_exists(window_title: str) -> bool: + """Checks if a window exists""" + if FindWindow(None, window_title) == 0: + return False + return True + + +def get_window_size(window_title: str) -> tuple: + """Gets the size of an open window""" + window_handle = FindWindow(None, window_title) + if window_handle == 0: + raise WindowNotFound + window_rect = GetWindowRect(window_handle) + return window_rect[0], window_rect[1], window_rect[2], window_rect[3] + + +def activate_windw(window_title: str) -> bool: + """Makes window the active window""" + try: + from pywinauto import Application + window = gw.getWindowsWithTitle(window_title) + if window: + window = window[0] # Get the first matching window + app = Application().connect(handle=window._hWnd) + app.window(handle=window._hWnd).set_focus() + return True + else: + print("Window not found!") + return False + except Exception as e: + print(f"Error on Windows: {e}") + return False diff --git a/lolbot/common/account.py b/lolbot/common/account.py deleted file mode 100644 index 094e37a..0000000 --- a/lolbot/common/account.py +++ /dev/null @@ -1,119 +0,0 @@ -""" -Handles accounts for the bot, persists to a JSON file -""" - -import os -import json -from abc import ABC, abstractmethod -from dataclasses import dataclass, asdict - -from lolbot.common.config import Constants - - -@dataclass -class Account: - username: str - password: str - level: int - - -class AccountGenerator(ABC): - - @abstractmethod - def get_account(self) -> Account: - pass - - @abstractmethod - def get_all_accounts(self) -> list: - pass - - @abstractmethod - def add_account(self): - pass - - @abstractmethod - def edit_account(self): - pass - - @abstractmethod - def delete_account(self): - pass - - @abstractmethod - def set_account_as_leveled(self): - pass - - -class AccountManager(AccountGenerator): - """Class that handles account persistence""" - - def __init__(self): - self.default_data = {'accounts': []} - if not os.path.exists(Constants.ACCOUNT_PATH): - with open(Constants.ACCOUNT_PATH, 'w+') as f: - json.dump(self.default_data, f, indent=4) - try: - with open(Constants.ACCOUNT_PATH, 'r') as f: - json.load(f) - except: - with open(Constants.ACCOUNT_PATH, 'w') as f: - json.dump(self.default_data, f, indent=4) - - def get_account(self, max_level: int) -> Account: - """Gets an account username from JSON file where level is < max_level""" - with open(Constants.ACCOUNT_PATH, "r") as f: - data = json.load(f) - for account in data['accounts']: - if account['level'] < max_level: - return Account(account['username'], account['password'], account['level']) - return Account('', '', 0) - - def add_account(self, account: Account) -> None: - """Writes account to JSON, will not write duplicates""" - with open(Constants.ACCOUNT_PATH, 'r+') as f: - data = json.load(f) - if asdict(account) in data['accounts']: - return - data['accounts'].append(asdict(account)) - with open(Constants.ACCOUNT_PATH, 'r+') as outfile: - json.dump(data, outfile, indent=4) - - def edit_account(self, og_uname: str, account: Account) -> None: - """Edit an account""" - with open(Constants.ACCOUNT_PATH, 'r') as f: - data = json.load(f) - index = -1 - for i in range(len(data['accounts'])): - if data['accounts'][i]['username'] == og_uname: - index = i - break - data['accounts'][index]['username'] = account.username - data['accounts'][index]['password'] = account.password - data['accounts'][index]['level'] = account.level - with open(Constants.ACCOUNT_PATH, 'w') as outfile: - json.dump(data, outfile, indent=4) - - def delete_account(self, account: Account) -> None: - """Deletes account""" - with open(Constants.ACCOUNT_PATH, 'r') as f: - data = json.load(f) - data['accounts'].remove(asdict(account)) - with open(Constants.ACCOUNT_PATH, 'w') as outfile: - json.dump(data, outfile, indent=4) - - def get_all_accounts(self) -> list: - """Returns all accounts as dictionary""" - with open(Constants.ACCOUNT_PATH, 'r') as f: - data = json.load(f) - return data['accounts'] - - def set_account_as_leveled(self, account: Account, max_level: int) -> None: - """Sets account level to user configured max level in the JSON file""" - with open(Constants.ACCOUNT_PATH, 'r') as f: - data = json.load(f) - for _account in data['accounts']: - if _account['username'] == account.username: - _account['level'] = max_level - with open(Constants.ACCOUNT_PATH, 'w') as outfile: - json.dump(data, outfile, indent=4) - return diff --git a/lolbot/common/accounts.py b/lolbot/common/accounts.py new file mode 100644 index 0000000..2bf1199 --- /dev/null +++ b/lolbot/common/accounts.py @@ -0,0 +1,68 @@ +""" +Handles multi-platform creating/writing accounts to json file. +""" + +import os +import json + +from lolbot.common.config import ACCOUNT_PATH + + +def load_accounts() -> list: + """Loads all accounts from disk.""" + accounts = [] + if not os.path.exists(ACCOUNT_PATH): + with open(ACCOUNT_PATH, 'w') as account_file: + json.dump(accounts, account_file, indent=4) + try: + with open(ACCOUNT_PATH, 'r') as account_file: + accounts = json.load(account_file) + except json.JSONDecodeError as e: + return accounts + except KeyError as e: + return accounts + if "accounts" in accounts: # convert old format to new + accounts = accounts['accounts'] + with open(ACCOUNT_PATH, 'w') as account_file: + json.dump(accounts, account_file, indent=4) + return accounts + + +def get_account(max_level: int = 10000) -> dict: + """Return first account under max_level.""" + accounts = load_accounts() + for account in accounts: + if account['level'] < max_level: + return account + return {"username": "", "password": "", "level": 0} + + +def save_or_add(account: dict) -> None: + """If an account with this username already exists, update it. Otherwise, add account.""" + accounts = load_accounts() + with open(ACCOUNT_PATH, 'w') as account_file: + exists = False + for acc in accounts: + if acc['username'] == account['username']: + acc.update(account) + exists = True + if not exists: + accounts.append(account) + json.dump(accounts, account_file, indent=4) + + +def update(username: str, account: dict) -> None: + """Updates any field of an account based on the original username.""" + accounts = load_accounts() + with open(ACCOUNT_PATH, 'w') as account_file: + for acc in accounts: + if acc['username'] == username: + acc.update(account) + json.dump(accounts, account_file, indent=4) + + +def delete(username: str) -> None: + """Removes an account based on the username.""" + accounts = load_accounts() + with open(ACCOUNT_PATH, 'w') as account_file: + json.dump([acc for acc in accounts if acc['username'] != username], account_file, indent=4) diff --git a/lolbot/common/api.py b/lolbot/common/api.py deleted file mode 100644 index 27ec4e3..0000000 --- a/lolbot/common/api.py +++ /dev/null @@ -1,147 +0,0 @@ -""" -Handles HTTP Requests for Riot Client and League Client -""" - -import logging -from base64 import b64encode -from time import sleep - -import requests -import urllib3 - -import lolbot.common.config as config - - -class Connection: - """Handles HTTP requests for Riot Client and League Client""" - - LCU_HOST = '127.0.0.1' - RCU_HOST = '127.0.0.1' - LCU_USERNAME = 'riot' - RCU_USERNAME = 'riot' - - def __init__(self) -> None: - self.client_type = '' - self.client_username = '' - self.client_password = '' - self.procname = '' - self.pid = '' - self.host = '' - self.port = '' - self.protocol = '' - self.headers = '' - self.session = requests.session() - self.config = config.ConfigRW() - self.log = logging.getLogger(__name__) - logging.getLogger('urllib3').setLevel(logging.INFO) - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - def set_rc_headers(self) -> None: - """Sets header info for Riot Client""" - self.log.debug("Initializing Riot Client session") - self.host = Connection.RCU_HOST - self.client_username = Connection.RCU_USERNAME - - # lockfile - lockfile = open(config.Constants.RIOT_LOCKFILE, 'r') - data = lockfile.read() - self.log.debug(data) - lockfile.close() - data = data.split(':') - self.procname = data[0] - self.pid = data[1] - self.port = data[2] - self.client_password = data[3] - self.protocol = data[4] - - # headers - userpass = b64encode(bytes('{}:{}'.format(self.client_username, self.client_password), 'utf-8')).decode('ascii') - self.headers = {'Authorization': 'Basic {}'.format(userpass), "Content-Type": "application/json"} - self.log.debug(self.headers['Authorization']) - - def set_lcu_headers(self, verbose: bool = True) -> None: - """Sets header info for League Client""" - self.host = Connection.LCU_HOST - self.client_username = Connection.LCU_USERNAME - - # lockfile - lockfile = open(self.config.get_data('league_lockfile'), 'r') - data = lockfile.read() - self.log.debug(data) - lockfile.close() - data = data.split(':') - self.procname = data[0] - self.pid = data[1] - self.port = data[2] - self.client_password = data[3] - self.protocol = data[4] - - # headers - userpass = b64encode(bytes('{}:{}'.format(self.client_username, self.client_password), 'utf-8')).decode('ascii') - self.headers = {'Authorization': 'Basic {}'.format(userpass)} - self.log.debug(self.headers['Authorization']) - - def connect_lcu(self, verbose: bool = True) -> None: - """Tries to connect to league client""" - if verbose: - self.log.info("Connecting to LCU API") - else: - self.log.debug("Connecting to LCU API") - self.host = Connection.LCU_HOST - self.client_username = Connection.LCU_USERNAME - - # lockfile - lockfile = open(self.config.get_data('league_lockfile'), 'r') - data = lockfile.read() - self.log.debug(data) - lockfile.close() - data = data.split(':') - self.procname = data[0] - self.pid = data[1] - self.port = data[2] - self.client_password = data[3] - self.protocol = data[4] - - # headers - userpass = b64encode(bytes('{}:{}'.format(self.client_username, self.client_password), 'utf-8')).decode('ascii') - self.headers = {'Authorization': 'Basic {}'.format(userpass)} - self.log.debug(self.headers['Authorization']) - - # connect - for i in range(30): - sleep(1) - try: - r = self.request('get', '/lol-login/v1/session') - except: - continue - if r.json()['state'] == 'SUCCEEDED': - if verbose: - self.log.info("Connection Successful") - else: - self.log.debug("Connection Successful") - self.request('post', '/lol-login/v1/delete-rso-on-close') # ensures self.logout after close - sleep(2) - return - raise Exception("Could not connect to League Client") - - def request(self, method: str, path: str, query: str = '', data: dict = None) -> requests.models.Response: - """Handles HTTP requests to Riot Client or League Client server""" - if data is None: - data = {} - if not query: - url = "{}://{}:{}{}".format(self.protocol, self.host, self.port, path) - else: - url = "{}://{}:{}{}?{}".format(self.protocol, self.host, self.port, path, query) - - if 'username' not in data: - self.log.debug("{} {} {}".format(method.upper(), url, data)) - else: - self.log.debug("{} {}".format(method.upper(), url)) - - fn = getattr(self.session, method) - - if not data: - r = fn(url, verify=False, headers=self.headers) - else: - r = fn(url, verify=False, headers=self.headers, json=data) - return r diff --git a/lolbot/common/config.py b/lolbot/common/config.py index d379d7b..d15369a 100644 --- a/lolbot/common/config.py +++ b/lolbot/common/config.py @@ -1,100 +1,68 @@ """ -Handles creating/writing configurations to json file +Handles multi-platform creating/writing LoLBot's configurations to json file. """ import os import json -from typing import Any - -class Constants: - """Constant settings""" - # Constant paths - RIOT_LOCKFILE = os.path.join(os.getenv('LOCALAPPDATA'), 'Riot Games/Riot Client/Config/lockfile') +if os.name == 'nt': CONFIG_DIR = os.path.join(os.getenv('LOCALAPPDATA'), 'LoLBot') - BAK_DIR = os.path.join(CONFIG_DIR, 'bak') - LOG_DIR = os.path.join(CONFIG_DIR, 'logs') - CONFIG_PATH = os.path.join(CONFIG_DIR, 'configs.json') - ACCOUNT_PATH = os.path.join(CONFIG_DIR, 'accounts.json') - - # Pyinstaller dependant paths - GAME_CFG = 'lolbot/resources/game.cfg' - ICON_PATH = 'lolbot/resources/images/a.ico' - - # Other - VERSION = '2.3.0' - - @staticmethod - def create_dirs(): - if not os.path.exists(Constants.CONFIG_DIR): - os.makedirs(Constants.CONFIG_DIR) - if not os.path.exists(Constants.LOG_DIR): - os.makedirs(Constants.LOG_DIR) - if not os.path.exists(Constants.BAK_DIR): - os.makedirs(Constants.BAK_DIR) - if not os.path.exists(Constants.CONFIG_PATH): - open(Constants.CONFIG_PATH, 'w+') - if not os.path.exists(Constants.ACCOUNT_PATH): - open(Constants.ACCOUNT_PATH, 'w+') - - -class DefaultSettings: - """Default settings for bot""" - LEAGUE_DIR = 'C:/Riot Games/League of Legends' - LOBBY = 880 - MAX_LEVEL = 30 - PATCH = '13.21.1' - CHAMPS = [21, 18, 22, 67] - DIALOG = ["mid ples", "plannin on goin mid team", "mid por favor", "bienvenidos, mid", "howdy, mid", "goin mid", "mid"] - - -class ConfigRW: - """Reads/Writes configurations required by bot""" - - def __init__(self): - self.file = open(Constants.CONFIG_PATH, 'r+') - self.settings = {} - self.load_or_default() - self._json_update() - - def _json_update(self): - """Persists settings""" - self.file.seek(0) - json.dump(self.settings, self.file, indent=4) - self.file.truncate() - - def load_or_default(self): - """Attempts to load settings, if it fails, sets to default""" +else: + CONFIG_DIR = os.path.join(os.path.expanduser('~'), 'Library', 'Application Support', 'LoLBot') + +os.makedirs(CONFIG_DIR, exist_ok=True) + +BAK_DIR = os.path.join(CONFIG_DIR, 'bak') +LOG_DIR = os.path.join(CONFIG_DIR, 'logs') +CONFIG_PATH = os.path.join(CONFIG_DIR, 'config.json') +ACCOUNT_PATH = os.path.join(CONFIG_DIR, 'accounts.json') + +GAME_CFG = 'lolbot/resources/game.cfg' + +ALL_LOBBIES = { + 'Draft Pick': 400, + 'Ranked Solo/Duo': 420, + 'Blind Pick': 430, + 'Ranked Flex': 440, + 'ARAM': 450, + 'Intro Bots': 870, + 'Beginner Bots': 880, + 'Intermediate Bots': 890, + 'Normal TFT': 1090, + 'Ranked TFT': 1100, + 'Hyper Roll TFT': 1130, + 'Double Up TFT': 1160 +} + +BOT_LOBBIES = { + 'Intro Bots': 870, + 'Beginner Bots': 880, + 'Intermediate Bots': 890, +} + + +def load_config() -> dict: + """Load configuration from disk or set defaults""" + default_config = { + 'league_dir': 'C:/Riot Games/League of Legends', + 'lobby': 880, + 'max_level': 30, + 'champs': [21, 18, 22, 67], + 'dialog': ["mid ples", "plannin on goin mid team", "mid por favor", "bienvenidos, mid", "howdy, mid", "goin mid", "mid"] + } + if not os.path.exists(CONFIG_PATH): + with open(CONFIG_PATH, 'w') as configfile: + json.dump(default_config, configfile, indent=4) + return default_config + else: try: - self.settings = json.load(self.file) - except json.decoder.JSONDecodeError: - self.set_defaults() - - def set_defaults(self): - """Set and persist the default settings""" - self.set_league_dir(DefaultSettings.LEAGUE_DIR) - self.settings['lobby'] = DefaultSettings.LOBBY - self.settings['max_level'] = DefaultSettings.MAX_LEVEL - self.settings['patch'] = DefaultSettings.PATCH - self.settings['champs'] = DefaultSettings.CHAMPS - self.settings['dialog'] = DefaultSettings.DIALOG - self._json_update() - - def set_league_dir(self, league_dir: str): - """Sets all league paths since they depend on one directory""" - self.settings['league_dir'] = league_dir - self.settings['league_path'] = os.path.join(league_dir, 'LeagueClient') - self.settings['league_config'] = os.path.join(league_dir, 'Config/game.cfg') - self.settings['league_lockfile'] = os.path.join(league_dir, 'lockfile') - self._json_update() + with open(CONFIG_PATH, 'r') as configfile: + return json.load(configfile) + except json.JSONDecodeError: + return default_config - def set_data(self, key: str, value: Any): - """Persists data to JSON""" - self.settings[key] = value - self._json_update() - def get_data(self, key: str): - """Retrieves data from JSON""" - for sk, sv in self.settings.items(): - if sk == key: - return self.settings[key] +def save_config(config) -> None: + """Save the configuration to disk""" + with open(CONFIG_PATH, 'w') as configfile: + json.dump(config, configfile, indent=4) diff --git a/lolbot/common/proc.py b/lolbot/common/proc.py new file mode 100644 index 0000000..97e80e2 --- /dev/null +++ b/lolbot/common/proc.py @@ -0,0 +1,92 @@ +""" +Utility functions that interact processes. +""" + +import logging +import subprocess +import os +import sys +from time import sleep + +log = logging.getLogger(__name__) + +# WINDOW NAMES +LEAGUE_CLIENT_WINNAME = "League of Legends" +LEAGUE_GAME_CLIENT_WINNAME = "League of Legends (TM) Client" + +# PROCESS NAMES +LEAGUE_PROCESS_NAMES = ["LeagueClient.exe", "League of Legends.exe"] +RIOT_CLIENT_PROCESS_NAMES = ["RiotClientUx.exe", "RiotClientServices.exe", "Riot Client.exe"] + +# COMMANDS +KILL_CRASH_HANDLER = 'TASKKILL /F /IM LeagueCrashHandler64.exe' +KILL_LEAGUE_CLIENT = 'TASKKILL /F /IM LeagueClient.exe' +KILL_LEAGUE = 'TASKKILL /F /IM "League of Legends.exe"' +KILL_RIOT_CLIENT = 'TASKKILL /F /IM RiotClientUx.exe' +KILL_HANDLER_WMIC = 'wmic process where "name=\'LeagueCrashHandler64.exe\'" delete' +KILL_LEAGUE_WMIC = 'wmic process where "name=\'LeagueClient.exe\'" delete' + + +def is_league_running() -> bool: + """Checks if league processes exists""" + res = subprocess.check_output(["TASKLIST"], creationflags=0x08000000) + output = str(res) + for name in LEAGUE_PROCESS_NAMES: + if name in output: + return True + return False + + +def is_rc_running() -> bool: + """Checks if riot client process exists""" + res = subprocess.check_output(["TASKLIST"], creationflags=0x08000000) + output = str(res) + for name in RIOT_CLIENT_PROCESS_NAMES: + if name in output: + return True + return False + + +def is_game_running() -> bool: + """Checks if game process exists""" + res = subprocess.check_output(["TASKLIST"], creationflags=0x08000000) + output = str(res) + if "League of Legends.exe" in output: + return True + return False + + +def close_all_processes() -> None: + """Closes all league related processes""" + log.info("Terminating league related processes") + os.system(KILL_CRASH_HANDLER) + os.system(KILL_LEAGUE) + os.system(KILL_LEAGUE_CLIENT) + os.system(KILL_RIOT_CLIENT) + os.system(KILL_HANDLER_WMIC) + os.system(KILL_LEAGUE_WMIC) + sleep(5) + + +def close_game() -> None: + """Closes the League of Legends game process""" + log.info("Terminating game instance") + os.system(KILL_LEAGUE) + sleep(15) + + +def resource_path(relative_path: str) -> str: + """Returns pyinstaller path if exe or abs path""" + if hasattr(sys, '_MEIPASS'): + return os.path.join(sys._MEIPASS, relative_path) + return os.path.join(os.path.abspath("."), relative_path) + + +def close_riot_client() -> None: + """Closes the League of Legends game process""" + log.info("Closing riot client") + try: + os.system(KILL_RIOT_CLIENT) + except: + log.warning("Could not kill riot client") + sleep(2) diff --git a/lolbot/common/utils.py b/lolbot/common/utils.py deleted file mode 100644 index 53876e5..0000000 --- a/lolbot/common/utils.py +++ /dev/null @@ -1,236 +0,0 @@ -""" -Utility functions that interact with game windows and processes -""" - -import logging -import subprocess -import os -import sys -from time import sleep - -import keyboard -import mouse -import pyautogui -from win32gui import FindWindow, GetWindowRect - - -log = logging.getLogger(__name__) - -# WINDOW NAMES -LEAGUE_CLIENT_WINNAME = "League of Legends" -LEAGUE_GAME_CLIENT_WINNAME = "League of Legends (TM) Client" - -# PROCESS NAMES -LEAGUE_PROCESS_NAMES = ["LeagueClient.exe", "League of Legends.exe"] -RIOT_CLIENT_PROCESS_NAMES = ["RiotClientUx.exe", "RiotClientServices.exe", "Riot Client.exe"] - -# COMMANDS -KILL_CRASH_HANDLER = 'TASKKILL /F /IM LeagueCrashHandler64.exe' -KILL_LEAGUE_CLIENT = 'TASKKILL /F /IM LeagueClient.exe' -KILL_LEAGUE = 'TASKKILL /F /IM "League of Legends.exe"' -KILL_RIOT_CLIENT = 'TASKKILL /F /IM RiotClientUx.exe' -KILL_HANDLER_WMIC = 'wmic process where "name=\'LeagueCrashHandler64.exe\'" delete' -KILL_LEAGUE_WMIC = 'wmic process where "name=\'LeagueClient.exe\'" delete' - - -class WindowNotFound(Exception): - pass - - -def is_league_running() -> bool: - """Checks if league processes exists""" - res = subprocess.check_output(["TASKLIST"], creationflags=0x08000000) - output = str(res) - for name in LEAGUE_PROCESS_NAMES: - if name in output: - return True - return False - - -def is_rc_running() -> bool: - """Checks if riot client process exists""" - res = subprocess.check_output(["TASKLIST"], creationflags=0x08000000) - output = str(res) - for name in RIOT_CLIENT_PROCESS_NAMES: - if name in output: - return True - return False - - -def is_game_running() -> bool: - """Checks if game process exists""" - res = subprocess.check_output(["TASKLIST"], creationflags=0x08000000) - output = str(res) - if "League of Legends.exe" in output: - return True - return False - - -def close_all_processes() -> None: - """Closes all league related processes""" - log.info("Terminating league related processes") - os.system(KILL_CRASH_HANDLER) - os.system(KILL_LEAGUE) - os.system(KILL_LEAGUE_CLIENT) - os.system(KILL_RIOT_CLIENT) - os.system(KILL_HANDLER_WMIC) - os.system(KILL_LEAGUE_WMIC) - sleep(5) - - -def close_game() -> None: - """Closes the League of Legends game process""" - log.info("Terminating game instance") - os.system(KILL_LEAGUE) - sleep(15) - - -def resource_path(relative_path: str) -> str: - """Returns pyinstaller path if exe or abs path""" - if hasattr(sys, '_MEIPASS'): - return os.path.join(sys._MEIPASS, relative_path) - return os.path.join(os.path.abspath("."), relative_path) - - -def close_riot_client() -> None: - """Closes the League of Legends game process""" - log.info("Closing riot client") - try: - os.system(KILL_RIOT_CLIENT) - except: - log.warning("Could not kill riot client") - sleep(2) - - -def size(window_title: str = LEAGUE_CLIENT_WINNAME) -> tuple: - """Gets the size of an open window""" - window_handle = FindWindow(None, window_title) - if window_handle == 0: - raise WindowNotFound - window_rect = GetWindowRect(window_handle) - return window_rect[0], window_rect[1], window_rect[2], window_rect[3] - - -def exists(window_title: str) -> bool: - """Checks if a window exists""" - if FindWindow(None, window_title) == 0: - return False - return True - - -def click(ratio: tuple, expected_window_name: str = '', wait: int or float = 1) -> None: - """Makes a click in an open window""" - if expected_window_name != '' and not exists(expected_window_name): - log.debug("Cannot click on {}, {} does not exist".format(ratio, expected_window_name)) - raise WindowNotFound - elif expected_window_name != '': - window_name = expected_window_name - else: # check if game is running and default to game otherwise set window to league client - if exists(LEAGUE_GAME_CLIENT_WINNAME): - window_name = LEAGUE_GAME_CLIENT_WINNAME - elif exists(LEAGUE_CLIENT_WINNAME): - window_name = LEAGUE_CLIENT_WINNAME - else: - log.debug("Cannot click on {}, no available window".format(ratio)) - return - log.debug('Clicking on ratio {}: {}, {}. Waiting: {}'.format(ratio, ratio[0], ratio[1], wait)) - x, y, l, h = size(window_name) - updated_x = ((l - x) * ratio[0]) + x - updated_y = ((h - y) * ratio[1]) + y - pyautogui.moveTo(updated_x, updated_y) - sleep(.5) - mouse.click() # pyautogui clicks do not work with league/directx - sleep(wait) - - -def right_click(ratio: tuple, expected_window: str = '', wait: int or float = 1) -> None: - """Makes a right click in an open window""" - if expected_window != '' and not exists(expected_window): - log.debug("Cannot click on {}, {} does not exist".format(ratio, expected_window)) - raise WindowNotFound - elif expected_window != '': - window_name = expected_window - else: # check if game is running and default to game otherwise set window to league client - if exists(LEAGUE_GAME_CLIENT_WINNAME): - window_name = LEAGUE_GAME_CLIENT_WINNAME - elif exists(LEAGUE_CLIENT_WINNAME): - window_name = LEAGUE_CLIENT_WINNAME - else: - log.debug("Cannot click on {}, no available window".format(ratio)) - return - log.debug('Clicking on ratio {}: {}, {}. Waiting: {}'.format(ratio, ratio[0], ratio[1], wait)) - x, y, l, h = size(window_name) - updated_x = ((l - x) * ratio[0]) + x - updated_y = ((h - y) * ratio[1]) + y - pyautogui.moveTo(updated_x, updated_y) - sleep(.5) - mouse.right_click() # pyautogui clicks do not work with league/directx - sleep(wait) - - -def attack_move_click(ratio: tuple, wait: int or float = 1) -> None: - """Attack move clicks in an open League of Legends game window""" - if not exists(LEAGUE_GAME_CLIENT_WINNAME): - log.debug("Cannot attack move when game is not running") - raise WindowNotFound - log.debug('Attack Moving on ratio {}: {}, {}. Waiting: {}'.format(ratio, ratio[0], ratio[1], wait)) - x, y, l, h = size(LEAGUE_GAME_CLIENT_WINNAME) - updated_x = ((l - x) * ratio[0]) + x - updated_y = ((h - y) * ratio[1]) + y - pyautogui.moveTo(updated_x, updated_y) - sleep(.5) - keyboard.press('a') - sleep(.1) - mouse.click() - sleep(.1) - mouse.click() - keyboard.release('a') - sleep(wait) - - -def press(key: str, expected_window: str = '', wait: int or float = 1) -> None: - """Sends a keypress to a window""" - if expected_window != '' and not exists(expected_window): - log.debug("Cannot press {}, {} does not exist".format(key, expected_window)) - raise WindowNotFound - log.debug("Pressing key: {}. Waiting: {}".format(key, wait)) - keyboard.press_and_release(key) - sleep(wait) - - -def write(keys: str, expected_window: str = '', wait: int or float = 1) -> None: - """Sends a string of key presses to a window""" - if expected_window != '' and not exists(expected_window): - log.debug("Cannot type {}, {} does not exist".format(keys, expected_window)) - raise WindowNotFound - log.debug("Typewriting {}. Waiting: {}".format(keys, wait)) - pyautogui.typewrite(keys) - sleep(wait) - - -def seconds_to_min_sec(seconds: str or float or int) -> str: - """Converts League of Legends game time to minute:seconds format""" - try: - if isinstance(seconds, int) or isinstance(seconds, float): - if len(str(int(seconds % 60))) == 1: - return str(int(seconds / 60)) + ":0" + str(int(seconds % 60)) - else: - return str(int(seconds / 60)) + ":" + str(int(seconds % 60)) - elif isinstance(seconds, str): - seconds = float(seconds) - if len(str(int(seconds % 60))) == 1: - return str(int(seconds / 60)) + ":0" + str(int(seconds % 60)) - else: - return str(int(seconds / 60)) + ":" + str(int(seconds % 60)) - except ValueError: - return "XX:XX" - - -def print_ascii() -> None: - """Prints some ascii art""" - print("""\n\n - ──────▄▌▐▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▌ - ───▄▄██▌█ BEEP BEEP - ▄▄▄▌▐██▌█ -15 LP DELIVERY - ███████▌█▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▄▌ - ▀(⊙)▀▀▀▀▀▀▀(⊙)(⊙)▀▀▀▀▀▀▀▀▀▀(⊙)\n\n\t\t\tLoL Bot\n\n""") diff --git a/lolbot/lcu/cmd.py b/lolbot/lcu/cmd.py new file mode 100644 index 0000000..ced29c4 --- /dev/null +++ b/lolbot/lcu/cmd.py @@ -0,0 +1,54 @@ +""" +Get league of legends auth url from process info. +""" + +import re +from dataclasses import dataclass + +import psutil + + +LCU_PORT_KEY = "--app-port=" +LCU_TOKEN_KEY = "--remoting-auth-token=" + +PORT_REGEX = re.compile(r"--app-port=(\d+)") +TOKEN_REGEX = re.compile(r"--remoting-auth-token=(\S+)") + +LEAGUE_PROCESS = "LeagueClientUx.exe" +RIOT_CLIENT_PROCESS = "Riot Client.exe" + +@dataclass +class CommandLineOutput: + auth_url: str = "" + token: str = "" + port: str = "" + + +def get_commandline() -> CommandLineOutput: + """ + Retrieves the command line of the LeagueClientUx.exe or Riot Client process and + returns the relevant information + """ + try: + for proc in psutil.process_iter(['name', 'cmdline']): + if proc.info['name'] == LEAGUE_PROCESS: + cmdline = " ".join(proc.info['cmdline']) + return match_stdout(cmdline) + elif proc.info['name'] == RIOT_CLIENT_PROCESS and PORT_REGEX.search(str(proc.info['cmdline'])): + cmdline = " ".join(proc.info['cmdline']) + return match_stdout(cmdline) + return CommandLineOutput() + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess) as e: + raise e + + +def match_stdout(stdout: str) -> CommandLineOutput: + """Parses the command line string to extract port, token, and directory""" + port_match = PORT_REGEX.search(stdout) + port = port_match.group(1).replace(LCU_PORT_KEY, '') if port_match else "0" + + token_match = TOKEN_REGEX.search(stdout) + token = token_match.group(1).replace(LCU_TOKEN_KEY, '').replace('"', '') if token_match else "" + + auth_url = f"https://riot:{token}@127.0.0.1:{port}" + return CommandLineOutput(auth_url=auth_url, token=token, port=port) diff --git a/lolbot/lcu/game_api.py b/lolbot/lcu/game_api.py new file mode 100644 index 0000000..ccb070d --- /dev/null +++ b/lolbot/lcu/game_api.py @@ -0,0 +1,99 @@ +""" +Handles all HTTP requests to the local game server, +providing functions for interacting with various game endpoints. +""" + +import json + +import requests + +GAME_SERVER_URL = 'https://127.0.0.1:2999/liveclientdata/allgamedata' +GAME_PROCESS_NAME = "League of Legends.exe" + + +class GameAPIError(Exception): + pass + + +def is_connected() -> bool: + """Check if getting response from game server""" + try: + response = requests.get(GAME_SERVER_URL, timeout=10, verify=False) + response.raise_for_status() + return True + except requests.RequestException as e: + return False + + +def get_game_data() -> str: + """Retrieves game data from the local game server""" + try: + response = requests.get(GAME_SERVER_URL, timeout=10, verify=False) + response.raise_for_status() + return response.text + except Exception as e: + raise GameAPIError(f"Failed to get game data: {str(e)}") + + +def get_game_time() -> int: + """Gets current time in game""" + try: + json_string = get_game_data() + data = json.loads(json_string) + return int(data['gameData']['gameTime']) + except json.JSONDecodeError as e: + raise GameAPIError(f"Invalid JSON data: {e}") + except KeyError as e: + raise GameAPIError(f"Missing key in data: {e}") + except GameAPIError as e: + raise e + + +def get_formatted_time() -> str: + """Converts League of Legends game time to minute:seconds format""" + try: + seconds = int(get_game_time()) + if len(str(int(seconds % 60))) == 1: + return str(int(seconds / 60)) + ":0" + str(int(seconds % 60)) + else: + return str(int(seconds / 60)) + ":" + str(int(seconds % 60)) + except GameAPIError: + return "XX:XX" + except ValueError: + return "XX:XX" + + +def get_champ() -> str: + """Gets current champion being played in game""" + try: + json_string = get_game_data() + data = json.loads(json_string) + for player in data['allPlayers']: + if player['summonerName'] == data['activePlayer']['summonerName']: + return player['championName'] + return "" + except json.JSONDecodeError as e: + raise GameAPIError(f"Invalid JSON data: {e}") + except KeyError as e: + raise GameAPIError(f"Missing key in data: {e}") + except GameAPIError as e: + raise e + + +def is_dead() -> bool: + """Returns whether player is currently dead""" + try: + json_string = get_game_data() + data = json.loads(json_string) + + dead = False + for player in data['allPlayers']: + if player['summonerName'] == data['activePlayer']['summonerName']: + dead = bool(player['isDead']) + return dead + except json.JSONDecodeError as e: + raise GameAPIError(f"Invalid JSON data: {e}") + except KeyError as e: + raise GameAPIError(f"Missing key in data: {e}") + except GameAPIError as e: + raise e diff --git a/lolbot/lcu/lcu_api.py b/lolbot/lcu/lcu_api.py new file mode 100644 index 0000000..15ecdf5 --- /dev/null +++ b/lolbot/lcu/lcu_api.py @@ -0,0 +1,383 @@ +""" +Handles all HTTP request to the local LoL Client, +providing functions for interacting with various LoL endpoints. +""" +import threading + +import requests +import urllib3 + +from lolbot.lcu import cmd + +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + +class LCUError(Exception): + """Exception for LCU API errors""" + pass + + +class LCUApi: + + def __init__(self): + self.client = requests.Session() + self.client.verify = False + self.client.headers.update({'Accept': 'application/json'}) + self.client.timeout = 2 + self.client.trust_env = False + self.timer = None + self.endpoint = cmd.get_commandline().auth_url + + def update_auth(self): + self.endpoint = cmd.get_commandline().auth_url + + def update_auth_timer(self, timer: int = 5): + self.endpoint = cmd.get_commandline().auth_url + self.timer = threading.Timer(timer, self.update_auth_timer) + self.timer.start() + + def stop_timer(self): + self.timer.cancel() + self.timer = None + + def make_get_request(self, url): + url = f"{self.endpoint}{url}" + try: + response = self.client.get(url) + return response + except Exception as e: + raise e + + def make_post_request(self, url, body): + url = f"{self.endpoint}{url}" + try: + response = self.client.post(url, json=body) + return response + except Exception as e: + raise e + + def make_patch_request(self, url, body): + url = f"{self.endpoint}{url}" + try: + response = self.client.patch(url, json=body) + return response + except Exception as e: + raise e + + def make_delete_request(self, url, body): + url = f"{self.endpoint}{url}" + try: + response = self.client.delete(url, json=body) + return response + except Exception as e: + raise e + + def make_put_request(self, url, body): + url = f"{self.endpoint}{url}" + try: + response = self.client.put(url, json=body) + return response + except Exception as e: + raise e + + def get_display_name(self) -> str: + """Gets display name of logged in account""" + url = f"{self.endpoint}/lol-summoner/v1/current-summoner" + try: + response = self.client.get(url) + response.raise_for_status() + return response.json()['displayName'] + except requests.RequestException as e: + raise LCUError(f"Error retrieving display name: {e}") + + def get_summoner_level(self) -> int: + """Gets level logged in account""" + url = f"{self.endpoint}/lol-summoner/v1/current-summoner" + try: + response = self.client.get(url) + response.raise_for_status() + return int(response.json()['summonerLevel']) + except requests.RequestException as e: + raise LCUError(f"Error retrieving summoner level: {e}") + + def get_patch(self) -> str: + """Gets current patch""" + url = f"{self.endpoint}/lol-patch/v1/game-version" + try: + response = self.client.get(url) + response.raise_for_status() + return response.text[1:6] + except requests.RequestException as e: + raise LCUError(f"Error retrieving patch: {e}") + + def get_lobby_id(self) -> int: + """Gets name of current lobby""" + url = f"{self.endpoint}/lol-lobby/v2/lobby" + try: + response = self.client.get(url) + response.raise_for_status() + return int(response.json()['gameConfig']['queueId']) + except requests.RequestException as e: + raise LCUError(f"Error retrieving lobby ID: {e}") + + def restart_ux(self) -> None: + """Restarts league client ux""" + url = f"{self.endpoint}/riotclient/kill-and-restart-ux" + try: + response = self.client.post(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Error restarting UX: {e}") + + def access_token_exists(self) -> bool: + """Checks if access token exists""" + url = f"{self.endpoint}/rso-auth/v1/authorization" + try: + response = self.client.get(url) + response.raise_for_status() + return True + except requests.RequestException as e: + return False + + def get_dodge_timer(self) -> int: + """Gets dodge penalty timer""" + url = f"{self.endpoint}/lol-matchmaking/v1/search" + try: + response = self.client.get(url) + response.raise_for_status() + if len(response.json()['errors']) != 0: + return int(response.json()['errors'][0]['penaltyTimeRemaining']) + else: + return 0 + except requests.RequestException as e: + raise LCUError(f"Error getting dodge timer: {e}") + + def get_estimated_queue_time(self) -> int: + """Retrieves estimated queue time""" + url = f"{self.endpoint}/lol-matchmaking/v1/search" + try: + response = self.client.get(url) + response.raise_for_status() + return int(response.json()['estimatedQueueTime']) + except requests.RequestException as e: + raise LCUError(f"Error retrieving estimated queue time: {e}") + + def login(self, username: str, password: str) -> None: + """Logs into the Riot Client""" + url = f"{self.endpoint}/rso-auth/v2/authorizations" + body = {"clientId": "riot-client", 'trustLevels': ['always_trusted']} + try: + response = self.client.post(url, json=body) + response.raise_for_status() + print(response.json()) + except requests.RequestException as e: + print(f"1{e}") + raise LCUError(f"Error in login authorization: {e}") + + url = f"{self.endpoint}/rso-auth/v1/session/credentials" + body = {"username": username, "password": password, "persistLogin": False} + try: + response = self.client.put(url, json=body) + response.raise_for_status() + print(response.json()) + except requests.RequestException as e: + print(e) + raise LCUError(f"Invalid Username or Password: {e}") + + def launch_league_from_rc(self) -> None: + """Ensures that the account does not stay signed in after client exits""" + url = f"{self.endpoint}/product-launcher/v1/products/league_of_legends/patchlines/live" + try: + response = self.client.post(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to launch league from Riot Client: {e}") + + def logout_on_close(self) -> None: + """Ensures that the account does not stay signed in after client exits""" + url = f"{self.endpoint}/lol-login/v1/delete-rso-on-close" + try: + response = self.client.post(url) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + raise LCUError(f"Failed to prevent auto-signin next launch: {e}") + + def get_phase(self) -> str: + """Retrieves the League Client phase""" + url = f"{self.endpoint}/lol-gameflow/v1/gameflow-phase" + try: + response = self.client.get(url) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + raise LCUError(f"Failed to get game phase: {e}") + + def create_lobby(self, lobby_id: int) -> None: + """Creates a lobby for given lobby ID""" + url = f"{self.endpoint}/lol-lobby/v2/lobby" + try: + response = self.client.post(url, json={'queueId': lobby_id}) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to create lobby with id {lobby_id}: {e}") + + def start_matchmaking(self) -> None: + """Starts matchmaking for current lobby""" + url = f"{self.endpoint}/lol-lobby/v2/lobby/matchmaking/search" + try: + response = self.client.post(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to start matchmaking: {e}") + + def get_matchmaking_time(self) -> str: + """Gets current time in queue""" + url = f"{self.endpoint}/lol-matchmaking/v1/search" + try: + response = self.client.get(url) + response.raise_for_status() + return str(int(response.json()['timeInQueue'])) + except requests.RequestException as e: + raise LCUError(f"Could not get time in queue: {e}") + + def quit_matchmaking(self) -> None: + """Cancels matchmaking search""" + url = f"{self.endpoint}/lol-lobby/v2/lobby/matchmaking/search" + try: + response = self.client.delete(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Error cancelling matchmaking: {e}") + + def accept_match(self) -> None: + """Accepts the Ready Check""" + url = f"{self.endpoint}/lol-matchmaking/v1/ready-check/accept" + try: + response = self.client.post(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to accept match: {e}") + + def get_champ_select_data(self) -> dict: + """Gets champ select lobby data""" + url = f"{self.endpoint}/lol-champ-select/v1/session" + try: + response = self.client.get(url) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + raise LCUError(f"Error retrieving champ select information: {e}") + + def get_available_champion_ids(self) -> list: + """Hovers a champion in champ select""" + url = f"{self.endpoint}/lol-lobby-team-builder/champ-select/v1/pickable-champion-ids" + try: + response = self.client.get(url) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + raise LCUError(f"Error getting available champs {e}") + + def hover_champion(self, action_id: str, champion_id) -> None: + """Hovers a champion in champ select""" + url = f"{self.endpoint}/lol-champ-select/v1/session/actions/{action_id}" + data = {'championId': champion_id} + try: + response = self.client.patch(url, json=data) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Error hovering champion in champ select {e}") + + def lock_in_champion(self, action_id: str, champion_id) -> None: + """Locks in a champion in champ select""" + url = f"{self.endpoint}/lol-champ-select/v1/session/actions/{action_id}/complete" + data = {'championId': champion_id} + try: + response = self.client.post(url, json=data) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Error locking in champion {e}") + + def get_cs_time_remaining(self) -> str: + """Returns time remaining in Champ Select""" + url = f"{self.endpoint}/lol-champ-select/v1/session" + try: + response = self.client.get(url) + response.raise_for_status() + return response.json()['timer']['adjustedTimeLeftInPhase'] + except requests.RequestException as e: + raise LCUError(f"Error retrieving champ select information: {e}") + + def game_reconnect(self): + """Reconnects to active game""" + url = f"{self.endpoint}/lol-gameflow/v1/reconnect" + try: + response = self.client.post(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Could not reconnect to game: {e}") + + def play_again(self): + """Moves the League Client from endgame screen back to lobby""" + url = f"{self.endpoint}/lol-lobby/v2/play-again" + try: + response = self.client.post(url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Could not exit play-again screen: {e}") + + def is_client_patching(self) -> bool: + """Checks if the client is currently patching""" + url = f"{self.endpoint}/patcher/v1/products/league_of_legends/state" + try: + response = self.client.get(url) + response.raise_for_status() + print(response.json()) + return not response.json()['isUpToDate'] + except requests.RequestException as e: + return False + + def get_players_to_honor(self) -> list: + """Returns list of player summonerIds that can then be honored""" + url = f"{self.endpoint}/lol-honor-v2/v1/ballot" + try: + response = self.client.get(url) + response.raise_for_status() + return response.json()['eligibleAllies'] + except requests.RequestException as e: + raise LCUError(f"Failed to retrieve players to honor player: {e}") + + def honor_player(self, summoner_id: int) -> None: + """Honors player in post game screen""" + url = f"{self.endpoint}/lol-honor-v2/v1/honor-player" + try: + response = self.client.post(url, data={"summonerID": summoner_id}) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to honor player: {e}") + + def send_chat_message(self, msg: str) -> None: + """Sends a message to the chat window""" + open_chats_url = f"{self.endpoint}/lol-chat/v1/conversations" + send_chat_message_url = f"{self.endpoint}/lol-chat/v1/conversations/{msg}/messages" + try: + response = self.client.get(open_chats_url) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to send message, could not retrieve chats: {e}") + + chat_id = None + for conversation in response.json(): + if conversation['gameName'] != '' and conversation['gameTag'] != '': + continue + chat_id = conversation['id'] + if chat_id is None: + raise LCUError(f"Failed to send message: Chat ID is NULL") + + message = {"body": msg} + try: + response = self.client.post(send_chat_message_url, json=message) + response.raise_for_status() + except requests.RequestException as e: + raise LCUError(f"Failed to send message: {e}") diff --git a/lolbot/view/about_tab.py b/lolbot/view/about_tab.py index 8de2ec1..8682621 100644 --- a/lolbot/view/about_tab.py +++ b/lolbot/view/about_tab.py @@ -1,5 +1,5 @@ """ -View tab that displays information about the bot +View tab that displays information about the bot. """ import webbrowser @@ -7,7 +7,7 @@ import dearpygui.dearpygui as dpg -from lolbot.common.config import Constants +VERSION = '3.0.0' class AboutTab: @@ -15,7 +15,7 @@ class AboutTab: def __init__(self) -> None: response = requests.get("https://api.github.com/repos/iholston/lol-bot/releases/latest") - self.version = 'v' + Constants.VERSION + self.version = 'v' + VERSION self.latest_version = response.json()["name"] self.need_update = False if self.latest_version != self.version: @@ -47,14 +47,10 @@ def _notes_text() -> str: notes = "\t\t\t\t\t\t\t\t\tNotes\n" notes += """ -Version 2.1.0+ processes accounts differently. Accounts must unfortunately -be added through the accounts tab manually a single time to convert them to -the new format. After that it should work normally. - -Alternatively, you can copy the old 'lolbot/resources/account.json' file to -the new location (select 'Show in Finder' on account tab). If you do this, -make sure to find and replace all instances of 'leveled' with 'level' and -'true' with 30 (or your max level) and false with 0. +Due to the release of Vanguard, it is recommended to run the bot as a python +script. There has been a distinct increase in accounts being banned on Windows +since Vanguard release. The bot is currently being ported to MacOS where there +is no Vanguard. Running on MacOS is the best way to not get banned. If you have any problems create an issue on the github repo. """ diff --git a/lolbot/view/accounts_tab.py b/lolbot/view/accounts_tab.py index 7066d2c..18cea25 100644 --- a/lolbot/view/accounts_tab.py +++ b/lolbot/view/accounts_tab.py @@ -1,5 +1,5 @@ """ -View tab that handles creation/editing of accounts +View tab that handles creation/editing of accounts. """ import subprocess @@ -9,8 +9,7 @@ from typing import Any import dearpygui.dearpygui as dpg -from lolbot.common.config import Constants -from lolbot.common.account import Account, AccountManager +from lolbot.common import config, accounts class AccountsTab: @@ -18,12 +17,10 @@ class AccountsTab: def __init__(self) -> None: self.id = None - self.am = AccountManager() self.accounts = None self.accounts_table = None def create_tab(self, parent: int) -> None: - """Creates Accounts Tab""" with dpg.tab(label="Accounts", parent=parent) as self.id: dpg.add_text("Options") dpg.add_spacer() @@ -39,7 +36,7 @@ def create_tab(self, parent: int) -> None: dpg.add_button(label="Cancel", width=113, callback=lambda: dpg.configure_item("AccountSubmit", show=False)) with dpg.group(horizontal=True): dpg.add_button(label="Add New Account", width=184, callback=lambda: dpg.configure_item("AccountSubmit", show=True)) - dpg.add_button(label="Show in File Explorer", width=184, callback=lambda: subprocess.Popen('explorer /select, {}'.format(Constants.ACCOUNT_PATH))) + dpg.add_button(label="Show in File Explorer", width=184, callback=lambda: subprocess.Popen('explorer /select, {}'.format(config.ACCOUNT_PATH))) dpg.add_button(tag="BackupButton", label="Create Backup", width=184, callback=self.create_backup) with dpg.tooltip(dpg.last_item()): dpg.add_text("Creates a backup of the accounts.json file in the bak folder") @@ -53,10 +50,9 @@ def create_tab(self, parent: int) -> None: self.create_accounts_table() def create_accounts_table(self) -> None: - """Creates a table from account data""" if self.accounts_table is not None: dpg.delete_item(self.accounts_table) - self.accounts = self.am.get_all_accounts() + self.accounts = accounts.load_accounts() with dpg.group(parent=self.id) as self.accounts_table: with dpg.group(horizontal=True): dpg.add_input_text(default_value=" Username", width=147) @@ -78,16 +74,17 @@ def create_accounts_table(self) -> None: dpg.add_button(label="Delete", callback=self.delete_account_dialog, user_data=acc) def add_account(self) -> None: - """Adds a new account to accounts.json and updates view""" dpg.configure_item("AccountSubmit", show=False) - self.am.add_account(Account(dpg.get_value("UsernameField"), dpg.get_value("PasswordField"), dpg.get_value("LevelField"))) + account = {"username": dpg.get_value("UsernameField"), "password": dpg.get_value("PasswordField"), "level": dpg.get_value("LevelField")} + accounts.save_or_add(account) dpg.configure_item("UsernameField", default_value="") dpg.configure_item("PasswordField", default_value="") dpg.configure_item("LevelField", default_value=False) self.create_accounts_table() def edit_account(self, sender, app_data, user_data: Any) -> None: - self.am.edit_account(user_data, Account(dpg.get_value("EditUsernameField"), dpg.get_value("EditPasswordField"), dpg.get_value("EditLevelField"))) + account = {"username": dpg.get_value("EditUsernameField"), "password": dpg.get_value("EditPasswordField"), "level": dpg.get_value("EditLevelField")} + accounts.update(user_data, account) dpg.delete_item("EditAccount") self.create_accounts_table() @@ -101,7 +98,7 @@ def edit_account_dialog(self, sender, app_data, user_data: Any) -> None: dpg.add_button(label="Cancel", width=113, callback=lambda: dpg.delete_item("EditAccount")) def delete_account(self, sender, app_data, user_data: Any) -> None: - self.am.delete_account(Account(user_data['username'], user_data['password'], user_data['level'])) + accounts.delete(user_data['username']) dpg.delete_item("DeleteAccount") self.create_accounts_table() @@ -119,10 +116,10 @@ def delete_account_dialog(self, sender, app_data, user_data: Any) -> None: @staticmethod def create_backup(sender: int) -> None: bak = "{}{}".format(time.strftime("%Y%m%d-%H%M%S"), ".json") - shutil.copyfile(Constants.ACCOUNT_PATH, '{}/{}'.format(Constants.BAK_DIR, bak)) + shutil.copyfile(config.ACCOUNT_PATH, '{}/{}'.format(config.BAK_DIR, bak)) dpg.configure_item("BackupButton", label="Backup Created!") threading.Timer(1, lambda: dpg.configure_item("BackupButton", label="Create Backup")).start() @staticmethod - def copy_2_clipboard(sender: int): + def copy_2_clipboard(sender: int) -> None: subprocess.run("clip", text=True, input=dpg.get_item_label(sender)) diff --git a/lolbot/view/bot_tab.py b/lolbot/view/bot_tab.py index b2dbdc6..d07ec57 100644 --- a/lolbot/view/bot_tab.py +++ b/lolbot/view/bot_tab.py @@ -1,174 +1,183 @@ """ -View tab that handles bot controls and displays bot output +View tab that handles bot controls and displays bot output. """ import os import multiprocessing -import requests import threading -from time import sleep +import time +import datetime +import textwrap import dearpygui.dearpygui as dpg -from lolbot.common import utils, api -from lolbot.common.config import ConfigRW -from lolbot.bot.client import Client +from lolbot.common import config, proc +from lolbot.lcu.lcu_api import LCUApi, LCUError +import lolbot.lcu.game_api as game_api +from lolbot.bot.bot import Bot class BotTab: """Class that displays the BotTab and handles bot controls/output""" - def __init__(self, message_queue: multiprocessing.Queue, terminate: threading.Event) -> None: - self.message_queue = message_queue - self.connection = api.Connection() - self.lobbies = { - 'Draft Pick': 400, - 'Ranked Solo/Duo': 420, - 'Blind Pick': 430, - 'Ranked Flex': 440, - 'ARAM': 450, - 'Intro Bots': 870, - 'Beginner Bots': 880, - 'Intermediate Bots': 890, - 'Normal TFT': 1090, - 'Ranked TFT': 1100, - 'Hyper Roll TFT': 1130, - 'Double Up TFT': 1160 - } - self.config = ConfigRW() - self.terminate = terminate + def __init__(self, api: LCUApi): + self.message_queue = multiprocessing.Queue() + self.games_played = multiprocessing.Value('i', 0) + self.bot_errors = multiprocessing.Value('i', 0) + self.api = api + self.output_queue = [] + self.endpoint = None self.bot_thread = None + self.start_time = None def create_tab(self, parent) -> None: - """Creates Bot Tab""" with dpg.tab(label="Bot", parent=parent) as self.status_tab: dpg.add_spacer() dpg.add_text(default_value="Controls") with dpg.group(horizontal=True): - dpg.add_button(tag="StartButton", label='Start Bot', width=93, callback=self.start_bot) # width=136 + dpg.add_button(tag="StartStopButton", label='Start Bot', width=93, callback=self.start_stop_bot) # width=136 dpg.add_button(label="Clear Output", width=93, callback=lambda: self.message_queue.put("Clear")) - dpg.add_button(label="Restart UX", width=93, callback=self.ux_callback) - dpg.add_button(label="Close Client", width=93, callback=self.close_client_callback) + dpg.add_button(label="Restart UX", width=93, callback=self.restart_ux) + dpg.add_button(label="Close Client", width=93, callback=self.close_client) dpg.add_spacer() - dpg.add_text(default_value="Info") - dpg.add_input_text(tag="Info", readonly=True, multiline=True, default_value="Initializing...", height=72, width=568, tab_input=True) + with dpg.group(horizontal=True): + with dpg.group(): + dpg.add_text(default_value="Info") + dpg.add_input_text(tag="Info", readonly=True, multiline=True, default_value="Initializing...", height=72, width=280, tab_input=True) + with dpg.group(): + dpg.add_text(default_value="Bot") + dpg.add_input_text(tag="Bot", readonly=True, multiline=True, default_value="Initializing...", height=72, width=280, tab_input=True) dpg.add_spacer() dpg.add_text(default_value="Output") dpg.add_input_text(tag="Output", multiline=True, default_value="", height=162, width=568, enabled=False) - self.update_info_panel() - def start_bot(self) -> None: - """Starts bot process""" + def start_stop_bot(self) -> None: if self.bot_thread is None: - if not os.path.exists(self.config.get_data('league_dir')): + if not os.path.exists(config.load_config()['league_dir']): self.message_queue.put("Clear") self.message_queue.put("League Installation Path is Invalid. Update Path to START") return self.message_queue.put("Clear") - self.bot_thread = multiprocessing.Process(target=Client, args=(self.message_queue,)) + self.start_time = time.time() + self.bot_thread = multiprocessing.Process(target=Bot().run, args=(self.message_queue, self.games_played, self.bot_errors)) self.bot_thread.start() - dpg.configure_item("StartButton", label="Quit Bot") + dpg.configure_item("StartStopButton", label="Quit Bot") else: - dpg.configure_item("StartButton", label="Start Bot") + dpg.configure_item("StartStopButton", label="Start Bot") self.stop_bot() - def stop_bot(self) -> None: - """Stops bot process""" + def stop_bot(self): if self.bot_thread is not None: self.bot_thread.terminate() self.bot_thread.join() self.bot_thread = None self.message_queue.put("Bot Successfully Terminated") - def ux_callback(self) -> None: - """Sends restart ux request to api""" - if utils.is_league_running(): - self.connection.request('post', '/riotclient/kill-and-restart-ux') - sleep(1) - self.connection.set_lcu_headers() - else: + def restart_ux(self) -> None: + if not proc.is_league_running(): self.message_queue.put("Cannot restart UX, League is not running") + return + try: + self.api.restart_ux() + except LCUError: + pass - def close_client_callback(self) -> None: + def close_client(self) -> None: """Closes all league related processes""" self.message_queue.put('Closing League Processes') - threading.Thread(target=utils.close_all_processes).start() + threading.Thread(target=proc.close_all_processes).start() def update_info_panel(self) -> None: - """Updates info panel text""" - if not self.terminate.is_set() and not utils.is_league_running(): - dpg.configure_item("Info", default_value="League is not running") - else: - if not self.terminate.is_set() and not os.path.exists(self.config.get_data('league_dir')): - self.message_queue.put("Clear") - self.message_queue.put("League Installation Path is Invalid. Update Path") - if not self.terminate.is_set(): - threading.Timer(2, self.update_info_panel).start() - else: - self.stop_bot() - return - - _account = "" - phase = "" - league_patch = "" - game_time = "" - champ = "" - level = "" - try: - if not self.connection.headers: - self.connection.set_lcu_headers() - r = self.connection.request('get', '/lol-summoner/v1/current-summoner') - if r.status_code == 200: - _account = r.json()['displayName'] - level = str(r.json()['summonerLevel']) + " - " + str( - r.json()['percentCompleteForNextLevel']) + "% to next level" - r = self.connection.request('get', '/lol-gameflow/v1/gameflow-phase') - if r.status_code == 200: - phase = r.json() - if phase == 'None': - phase = "In Main Menu" - elif phase == 'Matchmaking': - phase = 'In Queue' - elif phase == 'Lobby': - r = self.connection.request('get', '/lol-lobby/v2/lobby') - for lobby, id in self.lobbies.items(): - if id == r.json()['gameConfig']['queueId']: - phase = lobby + ' Lobby' - except: - try: - self.connection.set_lcu_headers() - except: - pass - if utils.is_game_running() or phase == "InProgress": - try: - response = requests.get('https://127.0.0.1:2999/liveclientdata/allgamedata', timeout=10, verify=False) - if response.status_code == 200: - for player in response.json()['allPlayers']: - if player['summonerName'] == response.json()['activePlayer']['summonerName']: - champ = player['championName'] - game_time = utils.seconds_to_min_sec(response.json()['gameData']['gameTime']) - except: + if not proc.is_league_running(): + msg = textwrap.dedent("""\ + Phase: Closed + Accnt: - + Level: - + Time : - + Champ: -""") + dpg.configure_item("Info", default_value=msg) + return + try: + phase = self.api.get_phase() + game_time = "-" + champ = "-" + match phase: + case "None": + phase = "In Main Menu" + case "Matchmaking": + phase = "In Queue" + game_time = self.api.get_matchmaking_time() + case "Lobby": + lobby_id = self.api.get_lobby_id() + for lobby, id in config.ALL_LOBBIES.items(): + if id == lobby_id: + phase = lobby + " Lobby" + case "ChampSelect": + game_time = self.api.get_cs_time_remaining() + case "InProgress": + phase = "In Game" try: - self.connection.set_lcu_headers() + game_time = game_api.get_formatted_time() + champ = game_api.get_champ() except: pass - msg = "Accnt: {}\n".format(_account) - msg = msg + "Phase: {}\n".format(phase) - msg = msg + "Time : {}\n".format(game_time) - msg = msg + "Champ: {}\n".format(champ) - msg = msg + "Level: {}".format(level) - else: - try: - r = requests.get('http://ddragon.leagueoflegends.com/api/versions.json') - league_patch = r.json()[0] - except: + case _: pass - msg = "Accnt: {}\n".format(_account) - msg = msg + "Phase: {}\n".format(phase) - msg = msg + "Patch: {}\n".format(league_patch) - msg = msg + "Level: {}".format(level) - if not self.terminate.is_set(): - dpg.configure_item("Info", default_value=msg) + msg = textwrap.dedent(f"""\ + Phase: {phase} + Accnt: {self.api.get_display_name()} + Level: {self.api.get_summoner_level()} + Time : {game_time} + Champ: {champ}""") + dpg.configure_item("Info", default_value=msg) + except LCUError: + pass - if not self.terminate.is_set(): - threading.Timer(2, self.update_info_panel).start() + def update_bot_panel(self): + msg = "" + if self.bot_thread is None: + msg += textwrap.dedent("""\ + Status : Ready + RunTime: - + Games : - + Errors : - + Action : -""") + else: + run_time = datetime.timedelta(seconds=(time.time() - self.start_time)) + hours, remainder = divmod(run_time.seconds, 3600) + minutes, seconds = divmod(remainder, 60) + if run_time.days > 0: + time_since_start = f"{run_time.days} day, {hours:02}:{minutes:02}:{seconds:02}" + else: + time_since_start = f"{hours:02}:{minutes:02}:{seconds:02}" + if len(self.output_queue) > 0: + action = f"{self.output_queue[-1].split(']')[-1].strip()}" + else: + action = "-" + msg = textwrap.dedent(f"""\ + Status : Running + RunTime: {time_since_start} + Games : {self.games_played.value} + Errors : {self.bot_errors.value} + Action : {action}""") + dpg.configure_item("Bot", default_value=msg) + + def update_output_panel(self): + """Updates output panel with latest log messages.""" + if not self.message_queue.empty(): + display_msg = "" + self.output_queue.append(self.message_queue.get()) + if len(self.output_queue) > 12: + self.output_queue.pop(0) + for msg in self.output_queue: + if "Clear" in msg: + self.output_queue = [] + display_msg = "" + break + elif "INFO" not in msg and "ERROR" not in msg and "WARNING" not in msg: + display_msg += f'[{datetime.datetime.now().strftime("%H:%M:%S")}] [INFO ] {msg}\n' + else: + display_msg += msg + "\n" + if "Bot Successfully Terminated" in display_msg: + self.output_queue = [] + dpg.configure_item("Output", default_value=display_msg.strip()) diff --git a/lolbot/view/config_tab.py b/lolbot/view/config_tab.py index e66c7ef..bcd613d 100644 --- a/lolbot/view/config_tab.py +++ b/lolbot/view/config_tab.py @@ -1,5 +1,5 @@ """ -View tab that sets configurations for the bot +View tab that sets configurations for the bot. """ import webbrowser @@ -7,7 +7,7 @@ import dearpygui.dearpygui as dpg -from lolbot.common.config import ConfigRW +import lolbot.common.config as config class ConfigTab: @@ -15,12 +15,7 @@ class ConfigTab: def __init__(self) -> None: self.id = None - self.lobbies = { - 'Intro': 870, - 'Beginner': 880, - 'Intermediate': 890 - } - self.config = ConfigRW() + self.config = config.load_config() def create_tab(self, parent: int) -> None: """Creates Settings Tab""" @@ -33,23 +28,23 @@ def create_tab(self, parent: int) -> None: dpg.add_spacer() with dpg.group(horizontal=True): dpg.add_input_text(default_value='League Installation Path', width=180, enabled=False) - dpg.add_input_text(tag="LeaguePath", default_value=self.config.get_data('league_dir'), width=380, callback=self._set_dir) + dpg.add_input_text(tag="LeaguePath", default_value=self.config['league_dir'], width=380, callback=self.save_config) with dpg.group(horizontal=True): dpg.add_input_text(default_value='Game Mode', width=180, readonly=True) - lobby = int(self.config.get_data('lobby')) + lobby = int(self.config['lobby']) if lobby < 870: lobby += 40 - dpg.add_combo(tag="GameMode", items=list(self.lobbies.keys()), default_value=list(self.lobbies.keys())[ - list(self.lobbies.values()).index(lobby)], width=380, callback=self._set_mode) + dpg.add_combo(tag="GameMode", items=list(config.BOT_LOBBIES.keys()), default_value=list(config.BOT_LOBBIES.keys())[ + list(config.BOT_LOBBIES.values()).index(lobby)], width=380, callback=self.save_config) with dpg.group(horizontal=True): dpg.add_input_text(default_value='Account Max Level', width=180, enabled=False) - dpg.add_input_int(tag="MaxLevel", default_value=self.config.get_data('max_level'), min_value=0, step=1, width=380, callback=self._set_level) + dpg.add_input_int(tag="MaxLevel", default_value=self.config['max_level'], min_value=0, step=1, width=380, callback=self.save_config) with dpg.group(horizontal=True): dpg.add_input_text(default_value='Champ Pick Order', width=180, enabled=False) with dpg.tooltip(dpg.last_item()): dpg.add_text("If blank or if all champs are taken, the bot\nwill select a random free to play champion.\nAdd champs with a comma between each number.\nIt will autosave if valid.") - dpg.add_input_text(default_value=str(self.config.get_data('champs')).replace("[", "").replace("]", ""), width=334, callback=self._set_champs) - b = dpg.add_button(label="list", width=42, indent=526, callback=lambda: webbrowser.open('ddragon.leagueoflegends.com/cdn/{}/data/en_US/champion.json'.format(self.config.get_data('patch')))) + dpg.add_input_text(tag="Champs", default_value=str(self.config['champs']).replace("[", "").replace("]", ""), width=334, callback=self.save_config) + b = dpg.add_button(label="list", width=42, indent=526, callback=lambda: webbrowser.open('ddragon.leagueoflegends.com/cdn/{}/data/en_US/champion.json'.format('14.21'))) with dpg.tooltip(dpg.last_item()): dpg.add_text("Open ddragon.leagueoflegends.com in webbrowser") dpg.bind_item_theme(b, "__hyperlinkTheme") @@ -58,34 +53,16 @@ def create_tab(self, parent: int) -> None: with dpg.tooltip(dpg.last_item()): dpg.add_text("The bot will type a random phrase in the\nchamp select lobby. Each line is a phrase.\nIt will autosave.") x = "" - for dia in self.config.get_data('dialog'): + for dia in self.config['dialog']: x += dia.replace("'", "") + "\n" - dpg.add_input_text(default_value=x, width=380, multiline=True, height=215, callback=self._set_dialog) + dpg.add_input_text(tag="Dialog", default_value=x, width=380, multiline=True, height=215, callback=self.save_config) - def _set_dir(self, sender: int) -> None: - """Checks if directory exists and sets the Client Directory path""" - _dir = dpg.get_value(sender) # https://stackoverflow.com/questions/42861643/python-global-variable-modified-prior-to-multiprocessing-call-is-passed-as-ori - if os.path.exists(_dir): - self.config.set_league_dir(_dir) - - def _set_mode(self, sender: int) -> None: - """Sets the game mode""" - self.config.set_data('lobby', self.lobbies.get(dpg.get_value(sender))) - - def _set_level(self, sender: int) -> None: - """Sets account max level""" - self.config.set_data('max_level', dpg.get_value(sender)) - - def _set_champs(self, sender: int) -> None: - """Sets champ pick order""" - x = dpg.get_value(sender) - try: - champs = [int(s) for s in x.split(',')] - except ValueError: - dpg.configure_item(sender, default_value=str(self.config.get_data('champs')).replace("[", "").replace("]", "")) - return - self.config.set_data('champs', champs) - - def _set_dialog(self, sender: int) -> None: - """Sets dialog options""" - self.config.set_data('dialog', dpg.get_value(sender).strip().split("\n")) + def save_config(self): + if os.path.exists(dpg.get_value('LeaguePath')): + self.config['league_dir'] = dpg.get_value('LeaguePath') + self.config['lobby'] = config.BOT_LOBBIES.get(dpg.get_value('GameMode')) + self.config['max_level'] = dpg.get_value('MaxLevel') + champs = dpg.get_value('Champs') + self.config['champs'] = [int(s) for s in champs.split(',')] + self.config['dialog'] = dpg.get_value("Dialog").strip().split("\n") + config.save_config(self.config) diff --git a/lolbot/view/http_tab.py b/lolbot/view/http_tab.py index 139e752..26d17a0 100644 --- a/lolbot/view/http_tab.py +++ b/lolbot/view/http_tab.py @@ -1,5 +1,5 @@ """ -View tab that sends custom HTTP requests to LCU API +View tab that sends custom HTTP requests to LCU API. """ import webbrowser @@ -8,22 +8,21 @@ import dearpygui.dearpygui as dpg -from lolbot.common import api +from lolbot.lcu.lcu_api import LCUApi class HTTPTab: """Class that displays the HTTPTab and sends custom HTTP requests to the LCU API""" - def __init__(self) -> None: + def __init__(self, api: LCUApi) -> None: self.id = -1 - self.connection = api.Connection() - self.methods = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'] + self.api = api def create_tab(self, parent: int) -> None: """Creates the HTTPTab""" with dpg.tab(label="HTTP", parent=parent) as self.id: dpg.add_text("Method:") - dpg.add_combo(tag='Method', items=self.methods, default_value='GET', width=569) + dpg.add_combo(tag='Method', items=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'], default_value='GET', width=569) dpg.add_text("URL:") dpg.add_input_text(tag='URL', width=568) dpg.add_text("Body:") @@ -69,15 +68,22 @@ def format_json() -> None: def request(self) -> None: """Sends custom HTTP request to LCU API""" try: - self.connection.set_lcu_headers() - except FileNotFoundError: - dpg.configure_item('StatusOutput', label='418') - dpg.configure_item('ResponseOutput', default_value='League of Legends is not running') - return - try: - r = self.connection.request(dpg.get_value('Method').lower(), dpg.get_value('URL').strip(), data=dpg.get_value('Body').strip()) - dpg.configure_item('StatusOutput', label=r.status_code) - dpg.configure_item('ResponseOutput', default_value=json.dumps(r.json(), indent=4)) + response = None + url = dpg.get_value('URL').strip() + body = dpg.get_value('Body').strip() + match dpg.get_value('Method').lower(): + case 'get': + response = self.api.make_get_request(url) + case 'post': + response = self.api.make_post_request(url, body) + case 'delete': + response = self.api.make_delete_request(url, body) + case 'put': + response = self.api.make_put_request(url, body) + case 'patch': + response = self.api.make_patch_request(url, body) + dpg.configure_item('StatusOutput', label=response.status_code) + dpg.configure_item('ResponseOutput', default_value=json.dumps(response.json(), indent=4)) except Exception as e: dpg.configure_item('StatusOutput', label='418') - dpg.configure_item('ResponseOutput', default_value=e.__str__()) + dpg.configure_item('ResponseOutput', default_value=str(e)) diff --git a/lolbot/view/logs_tab.py b/lolbot/view/logs_tab.py index ec1093a..621b4f7 100644 --- a/lolbot/view/logs_tab.py +++ b/lolbot/view/logs_tab.py @@ -1,5 +1,5 @@ """ -View tab that displays logs in the /logs folder +View tab that displays logs in the /logs folder. """ import subprocess @@ -9,7 +9,7 @@ import dearpygui.dearpygui as dpg -from lolbot.common.config import Constants +import lolbot.common.config as config class LogsTab: @@ -20,7 +20,6 @@ def __init__(self) -> None: self.logs_group = None def create_tab(self, parent) -> None: - """Creates Log Tab""" with dpg.tab(label="Logs", parent=parent) as self.id: with dpg.window(label="Delete Files", modal=True, show=False, tag="DeleteFiles", pos=[115, 130]): dpg.add_text("All files in the logs folder will be deleted") @@ -36,7 +35,7 @@ def create_tab(self, parent) -> None: dpg.add_text(tag="LogUpdatedTime", default_value='Last Updated: {}'.format(datetime.now())) dpg.add_button(label='Update', width=54, callback=self.create_log_table) dpg.add_button(label='Clear', width=54, callback=lambda: dpg.configure_item("DeleteFiles", show=True)) - dpg.add_button(label='Show in File Explorer', callback=lambda: subprocess.Popen('explorer /select, {}'.format(Constants.LOG_DIR))) + dpg.add_button(label='Show in File Explorer', callback=lambda: subprocess.Popen('explorer /select, {}'.format(config.LOG_DIR))) dpg.add_spacer() dpg.add_separator() dpg.add_spacer() @@ -48,8 +47,8 @@ def create_log_table(self) -> None: dpg.delete_item(self.logs_group) dpg.set_value('LogUpdatedTime', 'Last Updated: {}'.format(datetime.now())) with dpg.group(parent=self.id) as self.logs_group: - for filename in self.sorted_dir_creation_time(Constants.LOG_DIR): - f = os.path.join(Constants.LOG_DIR, filename) + for filename in self.sorted_dir_creation_time(config.LOG_DIR): + f = os.path.join(config.LOG_DIR, filename) if f.endswith('.1'): os.unlink(f) continue @@ -61,7 +60,7 @@ def create_log_table(self) -> None: def clear_logs(self) -> None: """Empties the log folder""" dpg.configure_item("DeleteFiles", show=False) - folder = Constants.LOG_DIR + folder = config.LOG_DIR for filename in os.listdir(folder): file_path = os.path.join(folder, filename) try: diff --git a/lolbot/view/main_window.py b/lolbot/view/main_window.py index 6ad388a..3f03c73 100644 --- a/lolbot/view/main_window.py +++ b/lolbot/view/main_window.py @@ -1,18 +1,14 @@ """ -User interface module that contains the main window +Main window that displays all the tabs. """ import ctypes; ctypes.windll.shcore.SetProcessDpiAwareness(0) # This must be set before importing pyautogui/dpg -import threading -import datetime -import multiprocessing +import multiprocessing; multiprocessing.freeze_support() # https://stackoverflow.com/questions/24944558/pyinstaller-built-windows-exe-fails-with-multiprocessing +import time import dearpygui.dearpygui as dpg -from lolbot.common import api -from lolbot.common import utils -from lolbot.common.account import AccountManager -from lolbot.common.config import Constants +from lolbot.lcu.lcu_api import LCUApi from lolbot.view.bot_tab import BotTab from lolbot.view.accounts_tab import AccountsTab from lolbot.view.config_tab import ConfigTab @@ -20,29 +16,24 @@ from lolbot.view.logs_tab import LogsTab from lolbot.view.about_tab import AboutTab +ICON_PATH = 'lolbot/resources/images/a.ico' + class MainWindow: """Class that displays the view""" def __init__(self, width: int, height: int) -> None: - multiprocessing.freeze_support() # https://stackoverflow.com/questions/24944558/pyinstaller-built-windows-exe-fails-with-multiprocessing - Constants.create_dirs() - - self.account_manager = AccountManager() - self.accounts = self.account_manager.get_all_accounts() - self.message_queue = multiprocessing.Queue() - self.output_queue = [] - self.connection = api.Connection() self.width = width self.height = height - self.terminate = threading.Event() self.tab_bar = None - self.bot_tab = BotTab(self.message_queue, self.terminate) + self.api = LCUApi() + self.bot_tab = BotTab(self.api) self.accounts_tab = AccountsTab() self.config_tab = ConfigTab() - self.https_tab = HTTPTab() + self.https_tab = HTTPTab(self.api) self.logs_tab = LogsTab() self.about_tab = AboutTab() + self.api.update_auth_timer() def show(self) -> None: """Renders view""" @@ -54,47 +45,29 @@ def show(self) -> None: dpg.add_theme_color(dpg.mvThemeCol_ButtonActive, [0, 0, 0, 0]) dpg.add_theme_color(dpg.mvThemeCol_ButtonHovered, [29, 151, 236, 25]) dpg.add_theme_color(dpg.mvThemeCol_Text, [29, 151, 236]) - with dpg.tab_bar(callback=self._tab_selected) as self.tab_bar: + with dpg.tab_bar() as self.tab_bar: self.bot_tab.create_tab(self.tab_bar) self.accounts_tab.create_tab(self.tab_bar) self.config_tab.create_tab(self.tab_bar) self.https_tab.create_tab(self.tab_bar) self.logs_tab.create_tab(self.tab_bar) self.about_tab.create_tab(self.tab_bar) - dpg.create_viewport(title='LoL Bot', width=self.width, height=self.height, small_icon=utils.resource_path(Constants.ICON_PATH), resizable=False) + dpg.create_viewport(title='LoL Bot', width=self.width, height=self.height, small_icon=ICON_PATH, resizable=False) dpg.setup_dearpygui() dpg.show_viewport() dpg.set_primary_window('primary window', True) - dpg.set_exit_callback(self.bot_tab.stop_bot) + dpg.set_exit_callback(self.on_exit) + panel_update_time = time.time() while dpg.is_dearpygui_running(): - self._gui_updater() + current_time = time.time() + if current_time - panel_update_time >= 0.3: + self.bot_tab.update_bot_panel() + self.bot_tab.update_info_panel() + self.bot_tab.update_output_panel() + panel_update_time = current_time dpg.render_dearpygui_frame() - self.terminate.set() dpg.destroy_context() - def _tab_selected(self, sender, app_data, user_data) -> None: - """Callback for tab select""" - if app_data == self.logs_tab.id: - self.logs_tab.create_log_table() - if app_data == self.accounts_tab.id: - self.accounts_tab.create_accounts_table() - - def _gui_updater(self) -> None: - """Updates view each frame, displays up-to-date bot info""" - if not self.message_queue.empty(): - display_message = "" - self.output_queue.append(self.message_queue.get()) - if len(self.output_queue) > 12: - self.output_queue.pop(0) - for msg in self.output_queue: - if "Clear" in msg: - self.output_queue = [] - display_message = "" - break - elif "INFO" not in msg and "ERROR" not in msg and "WARNING" not in msg: - display_message += "[{}] [INFO ] {}\n".format(datetime.datetime.now().strftime("%H:%M:%S"), msg) - else: - display_message += msg + "\n" - dpg.configure_item("Output", default_value=display_message.strip()) - if "Bot Successfully Terminated" in display_message: - self.output_queue = [] + def on_exit(self): + self.api.stop_timer() + self.bot_tab.stop_bot() diff --git a/main.pyw b/main.pyw index 5dd34e8..798f07f 100644 --- a/main.pyw +++ b/main.pyw @@ -4,7 +4,6 @@ Where bot execution starts from lolbot.view.main_window import MainWindow - if __name__ == '__main__': gui: MainWindow = MainWindow(600, 420) gui.show() diff --git a/requirements.txt b/requirements.txt index 497c108..fab38b8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,5 @@ pywin32 requests urllib3 pyautogui -dearpygui \ No newline at end of file +dearpygui +psutil \ No newline at end of file