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))
