Reputation: 13
I've been trying to figure out how to use this library called steam so I can set up a dota 2 bot with the dota2 library in python so it can create and be in multiple games. That, aside from requiring me to create different instances which is its own problem, would require me to set up the sentry files so it doesnt need login for each object that tries to run but I couldn't figure out how.
I'm running this bot along with a Flask http server so it can receive requests from my backend and create the lobbys from there.
I tried the documentation's way which is not much but I've tried and nothing. Does anybody know how I could achieve this?
It is supposed to? generate a login_key for each succesfull login but it doesn't generate it
(I've been using client.login(username=os.getenv("STEAM_USERNAME"), password=os.getenv("STEAM_PASSWORD"))
for the logins)
I also found a method that stores sentrys binaries but I don't know what should be in there in the first place, the documentation doesn't seem to have anything on this.
from steam.client import SteamClient
from dota2.client import Dota2Client
from dota2.enums import DOTA_GC_TEAM, DOTA_GameState
import time
import os
from dotenv import load_dotenv
import logging
logger = logging.getLogger(__name__)
class BotInstance:
def __init__(self):
load_dotenv()
self.client = SteamClient()
sentry_path = os.path.abspath('./sentry')
os.makedirs(sentry_path, exist_ok=True)
self.client.set_credential_location(sentry_path)
self.dota = Dota2Client(self.client)
self.in_lobby = False
self.setup_handlers()
def setup_handlers(self):
@self.client.on(self.client.EVENT_AUTH_CODE_REQUIRED)
def _on_auth_code(is_2fa, code_mismatch):
if is_2fa:
code = input("2FA: ")
self.client.login(os.getenv("STEAM_USERNAME"),
os.getenv("STEAM_PASSWORD"),
two_factor_code=code)
else:
code = input("Email code: ")
self.client.login(os.getenv("STEAM_USERNAME"),
os.getenv("STEAM_PASSWORD"),
auth_code=code)
self.client.on('logged_on', self._on_logged_on)
self.client.on("error", lambda e: logger.error(f"Error: {e}"))
self.dota.on('ready', self._on_dota_ready)
self.dota.on("lobby_new", self._on_lobby_created)
self.dota.on("lobby_changed", self._on_lobby_changed)
self.dota.on("lobby_removed", self._on_lobby_removed)
def _on_logged_on(self):
if self.client.login_key:
with open(os.path.join(self.client.credential_location, f'sentry_{self.client.username.lower()}.key'), 'w') as f:
f.write(self.client.login_key)
self.dota.launch()
def _on_dota_ready(self):
if self.dota.lobby:
logger.warning("Exiting current lobby")
self.dota.leave_practice_lobby()
time.sleep(2)
logger.info("Dota 2 bot ready")
def _on_lobby_created(self, lobby):
logger.info(f"Lobby created: {lobby}")
self.in_lobby = True
def _on_lobby_changed(self, lobby):
logger.info("Lobby updated")
if self.dota.lobby:
logger.info("Current members in lobby: %s" % len(self.dota.lobby.all_members))
logger.info("State %s" % str(self.dota.lobby.state))
# if self.dota.lobby.state == DOTA_GameState.DOTA_GAMERULES_STATE_GAME_IN_PROGRESS:
# self.dota.leave_practice_lobby()
def _on_lobby_removed(self, lobby):
logger.info("Exiting lobby")
self.in_lobby = False
def check_players_in_lobby(self, steam_ids):
if not self.dota.lobby:
return False
lobby_members = [member.id for member in self.dota.lobby.all_members]
return all(int(steam_id) in lobby_members for steam_id in steam_ids)
def join_as_spectator(self):
try:
if not self.dota.lobby:
logger.warning("No lobby to join")
return
time.sleep(3)
self.dota.join_practice_lobby_team(team=DOTA_GC_TEAM.PLAYER_POOL)
self.in_lobby = True
logger.info("Joined as spectator")
except Exception as e:
logger.error("Error: %s" % e)
def create_and_go(self, name, password, game_mode, steam_ids):
try:
if self.in_lobby:
logger.warning("In another lobby")
return {'launch': 400, 'message': 'Bot is busy'}
logger.info(f"Creating lobby with pass: {password} with gamemode: {game_mode}")
self.dota.create_practice_lobby(password=password, options={
"game_name": name,
"game_mode": game_mode,
"allow_cheats": False,
"fill_with_bots": False,
"allow_spectating": True,
"server_region": 15,
})
wait_time = 10
start_time = time.time()
while not self.dota.lobby and time.time() - start_time < wait_time:
time.sleep(1)
if self.dota.lobby:
logger.info("Lobby created")
self.join_as_spectator()
if len(steam_ids) > 0:
for steam_id in steam_ids:
self.dota.invite_to_lobby(int(steam_id))
logger.info(f"Invited {steam_id}")
wait_time = 30
start_time = time.time()
while not self.check_players_in_lobby(steam_ids) and time.time() - start_time < wait_time:
time.sleep(1)
if self.check_players_in_lobby(steam_ids):
logger.info("All players in lobby")
logger.info("Lobby created")
time.sleep(3)
# self.dota.leave_practice_lobby()
self.dota.launch_practice_lobby()
# self.in_lobby = False
return {'launch': 200}
else:
logger.warning("Not all players in lobby")
self.dota.destroy_lobby()
self.in_lobby = False
return {'launch': 400, 'message': 'Not all players in lobby'}
else:
logger.warning("No steam ids to invite")
self.dota.destroy_lobby()
self.in_lobby = False
else:
logger.warning("Lobby not created")
return {'launch': 400, 'message': 'Lobby not created'}
except Exception as e:
logger.error(f"Error: {e}")
self.dota.leave_practice_lobby()
self.in_lobby = False
return {'launch': 400, 'message': str(e)}
def run(self):
try :
self.client.login(username=os.getenv("STEAM_USERNAME"), password=os.getenv("STEAM_PASSWORD"))
self.client.run_forever()
except Exception as e:
logger.error(f"Error: {e}")
self.client.logout()
def shutdown(self):
if self.dota.lobby:
self.dota.leave_practice_lobby()
self.in_lobby = False
self.client.logout()
self.client.disconnect()
Upvotes: 1
Views: 36