Script to automate image searching and clicking on screen

Question

For testing purposes of a platform I'm working on, I must automate the process of initiating Dota 2, accepting an invite that shows on screen, choose the right spot for that player (team A or team B), wait until the game starts and then exit it.

All of that could be done with simple clicks on static coordinates but Steam is not concerned about people trying to automate tests. So a lot can go wrong from the moment the game is initiate until the user arrives in the lobby. Update requests may appear in the screen for example, also, sometimes a fullscreen advertising may appear blocking the view of everything else, and they are different from each other so I can't rely on a 'x' close button on a specific place to click.

So, after many tries, I came up with this script below. I took several screenshots of every possible and obligatory step to do what I need, like a screenshot of the checkbox to "launch the game as soon as updated". Also, some coordinates are static and didn't need to be searched by an image reference. The script uses an image search library to find all these images at their specific moments. For example, before searching for the main logo of Dota, I have to search for the mentioned update screen that may appear. But because the update screen may not appear (in case it is already updated), I have to search for the main logo alongside, so if the main logo appears it means the game is not going to update, so this step is concluded and we can move forward.

This kind of thing happens a lot of times, it is similar to the idea of a fluxogram that can have multiple nodes "active" at once (by active I mean that they are being executed alongside each other).

So, the classes CoordNode and ImageSearchNode holds specific methods for each of the processess (the process of simply clicking on a coordinate, or searching for an image, and maybe clicking on it). Also, it serves to save information about the state of that process (if it's concluded or not, so things don't get clicked twice).

The script works as I expected sucessfully, but I'm not satisfied with the implementation.

Besides being a little bit unreadable from the function run_complete_flow, every modification or increase in functionality that has to be made takes too long to be understood on how to be implemented alongside the rest of the script.

Being a not so specific situation to happen (automate a process that handles several obligatory and non-obligatory events), I wonder if there's already a better standard implementation of such a job. Or at least some recommendation on how to organize better the code. Thank you.

import traceback
import logging
import subprocess
from image_search.imagesearch import imagesearch_num_loop, click_image, imagesearch_region_loop, save_screenshot
import sys
from get_tag_value import get_coords, get_info
import pyautogui
import psycopg2
import time
from telegram_logs.bot import telegram_logger, send_photo
import ctypes
import os
from private_module import get_node_5
logging.basicConfig(level=logging.INFO)
# powerbutton topleft corner should be at 987, 8
X_REF_LOGO = 987
Y_REF_LOGO = 8
X_CORRECTION = None
Y_CORRECTION = None
def basic_log(msg):
 telegram_logger(f"*{get_info('username')}*\n{str(msg)}")
def send_photolog():
 save_screenshot(get_info('username'))
 send_photo(f"*{get_info('username')}*", f"{get_info('username')}.png")
def run_game():
 if os.path.exists("tags.json"):
 os.remove("tags.json")
 detached_process = 0x00000008
 steam_username = get_info('username')
 password = get_info('password')
 print(steam_username, " : ", password)
 subprocess.Popen([r"C:\Program Files (x86)\Steam\Steam.exe", "-login", f"{steam_username}", f"{password}",
 "steam://rungameid/570"], creationflags=detached_process)
 user32 = ctypes.windll.user32
 screensize = user32.GetSystemMetrics(0), user32.GetSystemMetrics(1)
 basic_log(f"Launched game!\n Screen res is: {screensize[0]}x{screensize[1]}")
 return None
class Node:
 def __init__(self, node_name, event_info, send_telegram_log):
 self._node_name = node_name
 self._event_info = event_info
 self._send_telegram_log = send_telegram_log
 self._finished = False
 @property
 def node_name(self):
 return self._node_name
 def _set_finished(self, value: bool):
 self._finished = value
 if self._send_telegram_log and value:
 self._send_log()
 @property
 def finished(self):
 return self._finished
 @staticmethod
 def _click_on_coords(latency, pos):
 pyautogui.moveTo(pos[0], pos[1], latency)
 pyautogui.click(button="left")
 return pos
 def _send_log(self):
 basic_log(self.node_name)
 def reset(self):
 self._set_finished(False)
 def execute(self, *args):
 pass
class CoordNode(Node):
 def __init__(self, node_name, coords, send_telegram_log=True):
 super().__init__(node_name=node_name, event_info=coords, send_telegram_log=send_telegram_log)
 try:
 if not len(coords) == 2 \
 or not type(coords[0]) == int \
 or not type(coords[1]) == int:
 raise ValueError("What kind of coordinates are you giving me?")
 except Exception as exc:
 logging.debug(str(exc))
 raise exc
 self._coords = coords
 @property
 def coords(self):
 return self._coords
 def click(self, latency=2):
 if self.finished:
 return True
 self._set_finished(True)
 return self._click_on_coords(latency, self.coords)
 def execute(self, latency=2):
 return self.click(latency)
class ImageSearchNode(Node):
 def __init__(self, node_name, path: str, precision=0.8, send_telegram_log=True, clickable=False):
 super().__init__(node_name=node_name, event_info=path, send_telegram_log=send_telegram_log)
 try:
 test_path = path
 replaces = "_.\\/-:"
 for char in replaces:
 test_path = test_path.replace(char, '')
 if not test_path.isalnum():
 raise ValueError("What kind of image path are you trying to set?")
 except Exception as exc:
 logging.debug(str(exc))
 raise exc
 self._precision = precision
 self._path = path
 self.clickable = clickable
 self._found = False
 self._clicked = False
 @property
 def filename(self):
 if len(self.path.split('\\')) == 1: # In case we're running on linux
 return str(self.path.split('/')[-1])
 else:
 return str(self.path.split('\\')[-1])
 @property
 def found(self):
 return self._found
 @property
 def clicked(self):
 return self._clicked
 @property
 def path(self):
 return self._path
 def click(self, latency=2, frequency=None, duration=None):
 if self.finished:
 return True
 pos = self.find(frequency, duration)
 if not pos:
 return False
 click_image(image=self.path, pos=pos, action="left", timestamp=latency, offset=0)
 self._set_finished(True)
 return True
 def find(self, frequency=2, duration=6*60*60, area_search=False, area_coords=None):
 if self.finished:
 return True
 iterations = frequency * duration
 period = 1 / frequency
 if area_search:
 try:
 pos = imagesearch_region_loop(self.path, period, iterations,
 area_coords[0],
 area_coords[1],
 area_coords[2],
 area_coords[3],
 self._precision,
 self.filename)
 if pos[0] != -1:
 self._found = True
 logging.debug("FOUND AT " + str(pos))
 if not self.clickable:
 self._set_finished(True)
 return pos
 else:
 return None
 except Exception as exc:
 raise ValueError(f"Where are my area coords?? [x1, y1, x2, y2] \n {str(exc)}")
 pos = imagesearch_num_loop(self.path, period, iterations, self._precision, self.filename)
 if pos[0] != -1:
 self._found = True
 logging.debug("FOUND AT " + str(pos))
 if not self.clickable:
 self._set_finished(True)
 return pos
 else:
 logging.info(self.filename + "Not found!")
 return None
 def execute(self, find=False, find_and_click=None, latency=0, frequency=None, duration=6*60*60):
 if not find and not find_and_click:
 raise ValueError("Well, I must either find or find and click!")
 if find_and_click:
 return self.click(latency=latency, frequency=frequency, duration=duration)
 if find:
 return self.find(frequency, duration) 
def define_correction(main_logo_corner):
 global X_CORRECTION
 global Y_CORRECTION
 X_CORRECTION = main_logo_corner[0] - X_REF_LOGO
 Y_CORRECTION = main_logo_corner[1] - Y_REF_LOGO
def adjusted_coords(pos):
 if not pos:
 return pos
 x_corr = pos[0] + X_CORRECTION
 y_corr = pos[1] + Y_CORRECTION
 return [x_corr, y_corr]
def run_complete_flow(oss="windows"):
 windows_prefix = r"C:\Users\Administrator\Desktop\files_in_remote_ec2\python_scripts\images_to_search\\"
 linux_prefix = r"/home/user/Desktop/images/"
 if oss == "windows":
 prefix = windows_prefix
 else:
 prefix = linux_prefix
 # ~ : maybe will or maybe will not occur
 # ^ : certainly will occur
 # ^- : certainly will occur but condition has to be applied
 # 1) ~ launch_as_soon_as_updated -> (find and click)
 # 1.1) ~ restart steam -> (find and click)
 # 1.1.5) ^- launch dota 2 again if steam restarted -> (execute dota)
 # 1.2) ^ loading logo -> (find it)
 # 2) ~ advertising -> (press esc)
 # 3) ^ check if there is power button -> (find it)
 # 3.1) ~ check if it was in a lobby before -> (find and click)
 # 3.1.1) ^- if it was in a lobby, leave it -> (find and click)
 # 4) ^ accept invite -> (find and click)
 # 5) ^ in lobby get the correct slot and join it -> (click it)
 # 6) ^ wait for about 3 minutes until match begins -> (wait)
 # 7) ^- leave the match if you are radiant -> (nothing)
 # 7.1) ^ click the arrow to go to dashboard -> (click it)
 # 7.2) ^ leave game -> (click it)
 # 7.3) ^ confirm leave game -> (click it)
 # 7.4) ^ disconnect from match -> (click it)
 # 8) ^- continue if you are dire -> (click it)
 lasau = ImageSearchNode(node_name="Node 1: Launch as soon as updated",
 path=prefix+"1_launch_game_update.png",
 clickable=False,
 precision=0.6, send_telegram_log=False)
 resteam = ImageSearchNode(node_name="Node 1.1: Restart steam",
 path=prefix+"restart_steam.png",
 clickable=True,
 precision=0.6)
 # node_1_1_5 = "execute dota again"
 mainlogo = ImageSearchNode(node_name="Node 1.2: Located loading logo",
 path=prefix+"main_logo.png",
 clickable=True,
 precision=0.7)
 # node_2 = "press esc if advertising (after logo appears and not powerbutton"
 pwbttn = ImageSearchNode(node_name="Node 3: Found power button",
 path=prefix+"3_power_button.png",
 clickable=False,
 precision=0.7)
 wasinlbb = ImageSearchNode(node_name="Node 3.1: It was in a lobby!",
 path=prefix+"back_to_lobby.png",
 clickable=True,
 precision=0.6)
 lveprvslbb = ImageSearchNode(node_name="Node 3.1.1: leave the previous lobby",
 path=prefix+"leave_this_lobby.png",
 clickable=True,
 precision=0.6)
 accinv = ImageSearchNode(node_name="Node 4: Accept invite",
 path=prefix+"4_accept_invite.png",
 clickable=True,
 precision=0.8)
 # node_5 = "Defined later on"
 # STEP 0: LAUNCH GAME
 run_game()
 # Non, begin the screenmonitor flow
 # The logic must be:
 # If an option is uncertain to occur, we try it alongside all the subsequent uncertain events
 # Also, it must include at least the first obligatory event to occur and, if this event occurs, stop searching
 # for the previous uncertain ones:
 # in this case we only need to find the power button, not click it
 start_time = time.time()
 while not mainlogo.execute(find=True, frequency=2, duration=1):
 if lasau.execute(find=True, frequency=2, duration=1):
 # While it doesn't find the checked box, keep trying to click it
 # sometimes the screen glitches and the checkbox isn't clicked
 # the image is 145x15
 # the checkbox itself is at (0,0) - (15,15)
 # therefore, it must click (7,7) + chckbox_coords
 # after a while, check if finds the checkbox full.. if not, click it again
 count = 0
 while not ImageSearchNode("Checkbox clicked?", path=prefix+"checked_box.png",
 send_telegram_log=False).find(duration=1) and count < 5:
 lasau.reset()
 chckbox_coords = lasau.execute(find=True, frequency=2, duration=1)
 print(str(chckbox_coords) + "Checkbox coordinates!")
 if isinstance(chckbox_coords, tuple):
 CoordNode("Clicking checkbox..", [chckbox_coords[0]+7, chckbox_coords[1]+7]).execute(latency=0)
 count += 1
 ImageSearchNode(node_name="Desperately clicking on play game", path=prefix+"play_game.png",
 clickable=True, send_telegram_log=False, precision=0.5).execute(find_and_click=True,
 frequency=2, duration=1)
 if time.time()-start_time > int(get_info("timeout_photolog")):
 start_time = time.time()
 send_photolog()
 if resteam.execute(find_and_click=True, frequency=2, duration=1):
 # it clicked on restart steam, so load the game again
 run_game()
 while True:
 # Found the logo!
 # Now, look for the power button.. maybe be overridden by an advertising, so we click to close
 reincident = False
 while not (pwcoords:=pwbttn.execute(find=True, frequency=2, duration=1)):
 mainlogo.reset()
 if not mainlogo.execute(find=True, frequency=2, duration=1):
 # well, the logo isn't there, but neither the power button. It must be an advertising
 if reincident:
 basic_log("Go check if the advertising has changed")
 time.sleep(10)
 if (advcoords:=ImageSearchNode(node_name="Advertising", path=prefix+"advertising.png").execute(find=True,
 frequency=2,
 duration=2)):
 tl_adv = get_coords("advertising_coords")
 CoordNode("Clicking advertising", [advcoords[0]+tl_adv[0],
 advcoords[1]+tl_adv[1]]).execute()
 reincident = True
 ###### COORDINATES NEEDS ADJUSTMENTS FROM HERE #######
 # todo coordinates are not being correctly corrected. Sometimes the game screen tilts to another place
 define_correction(pwcoords)
 # Found power button! Just do a quick check on possible previous games
 if wasinlbb.execute(find_and_click=True, frequency=4, duration=4):
 lveprvslbb.execute(find_and_click=True, frequency=4, duration=5)
 # Now we are ready to accept invites!
 # Get the invitation!!
 while not accinv.execute(find_and_click=True, frequency=2):
 continue
 # now we can set node_5:
 coordinates, is_radiant = get_node_5() 
 coordinates = adjusted_coords(coordinates)
 # Click the correct spot
 node_5 = CoordNode(node_name="Node 5: found right spot",
 coords=coordinates)
 time.sleep(4) # just an assurance, we're running on slow computers
 while not node_5.execute():
 continue
 # sleep for 4min = 240s
 # node_6
 sleep_time = int(get_info('sleep_time'))
 logging.info(f"sleeping for {str(sleep_time)} seconds")
 time.sleep(sleep_time)
 if not is_radiant:
 # if is_radiant we proceed to leave the match, and let dire win:
 a_coords = get_coords('arrow_coords')
 d_coords = get_coords('disconnect_coords')
 l_coords = get_coords('leave_coords')
 c_coords = get_coords('confirm_coords')
 node_7_1 = CoordNode(node_name="Node 7.1 - clicked arrow", coords=adjusted_coords(a_coords))
 node_7_1.execute(latency=2)
 node_7_2 = CoordNode(node_name="Node 7.2 - clicked disconnect", coords=adjusted_coords(d_coords))
 node_7_2.execute(latency=2)
 node_7_3 = CoordNode(node_name="Node 7.3 - clicked leave", coords=adjusted_coords(l_coords))
 node_7_3.execute(latency=2)
 node_7_4 = CoordNode(node_name="Node 7.4 - clicked continue", coords=adjusted_coords(c_coords))
 node_7_4.execute(latency=2)
 else:
 cont_coords = get_coords('continue_coords')
 node_8 = CoordNode(node_name="Node 8 - clicked continue", coords=adjusted_coords(cont_coords))
 node_8.execute(latency=2)
 pwbttn.reset()
 wasinlbb.reset()
 lveprvslbb.reset()
 accinv.reset()
if __name__ == "__main__":
 if len(sys.argv) > 1:
 run_complete_flow("linux")
 else:
 try:
 run_complete_flow()
 except Exception:
 try:
 username = get_info('username')
 telegram_logger(f"*{username}* \n {str(traceback.format_exc())}")
 except Exception as ee:
 telegram_logger("\n\n second flaw \n\n" + str(ee))

• Ben A: What version of python are you using?
• Teodoro: @Linny, I'm using python 3.8

Answer 1

Have you considered to split up your run_complete_flow() function? Personally I would prefer to put it in a class. The first part up to run_game would go into __init__(). Then I would try to split up the task into methods. E.g. you have a comment # in this case we only need to find the power button, not click it - why not put that code in a method like find_power_buttom()? My goal would be, to make the flow easily understandable by reading the code of run_complete_flow() without comments, by "hiding all the unnecessary stuff" in well named methods. That should make it a lot easier to change the flow if needed, or identify where to look when a certain step makes problems.

def run_complete_flow(self):
 self.launch_as_soon_as_updated()
 self.advertising()
 self.check_power_button()
 self.accept_invite()
 self.in_lobby_get_slot_and_join()
 self.wait_for_match(time_out=180)
 radiant = self.check_radiant()
 if radiant:
 self.leave_match()
 else:
 # not sure, where to jump from here; consider puting whole 
 # radiant/dire "loop" in one method continue_until_radiant
 self.continue_if_dire() 
)

• Teodoro: Well, I haven't considered. It is a good start on organizing the code, but it doesn't actually solves my problem, since inside each of these functions there will still be the same block of code I would like to avoid or rebuild (the while's with confusing logic of when to break). But I will do this refactor for now, thank you :)