alex/my-botty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Breaking Messenger Module into something more flexible for specif… (#370)

Browse Source 
* WIP Breaking Messenger Module into something more flexible for specific apis

* Cleanup and removal of data class

* Working embeds sort of

* Simple embeds set up

* Adding image of stashing items to discord embed

* Discord Embed updates with images

death_manager and game_controller both store path to last death/chicken screenshot
Pass last screenshot path along to game_stats and discord_embeds to place in discord message

* Color adjust

* Updating filenames formobile notifications

* Create loot_screenshots folder if it doesn't exist and api=discord

* Fixing missed messenger.send calls

* UI for discord updates

* Switching to i_api and explicit functions

* Update all messenger calls to new functions

* Param name change

* Simplify discord_basic api

* Switching from interface to generic api class

* One new messenger call from master

* spacing
This commit is contained in:
egut125
2022-01-06 18:33:59 +01:00
committed by GitHub
parent b01ff94b8a
commit dff178bbc7
16 changed files with 246 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/params.ini
+1
View File
@@ -2,6 +2,7 @@
[general]
name=Botty
monitor=0
message_api_type=generic_api
custom_message_hook=
logg_lvl=debug
max_game_length_s=380
environment.yml
+1
View File
@@ -21,3 +21,4 @@ dependencies:
- graphviz
- psutil
- pillow
- discord.py
src/api/__init__.py
+2
View File
@@ -0,0 +1,2 @@
from .generic_api import GenericApi
from .discord_embeds import DiscordEmbeds
src/api/discord_embeds.py
+87
View File
@@ -0,0 +1,87 @@
from .generic_api import GenericApi
from config import Config
import cv2
import datetime
import discord
from version import __version__
import numpy as np
from discord import Webhook, RequestsWebhookAdapter, Color
class DiscordEmbeds(GenericApi):
def __init__(self):
self._config = Config()
self._webhook = Webhook.from_url(self._config.general['custom_message_hook'], adapter=RequestsWebhookAdapter(), )
self._file = None
self._psnURL = "https://i.psnprofiles.com/games/3bffee/trophies/"
def send_item(self, item: str, image: np.ndarray, location: str):
imgName = item.replace('_', '-')
cv2.imwrite(f"./loot_screenshots/{item}.png", image)
file = discord.File(f"./loot_screenshots/{item}.png", filename=f"{imgName}.png")
e = discord.Embed(
title="Item Stashed!",
description=f"{item} at {location}",
color=self._get_Item_Color( item),
)
e.set_thumbnail(url=f"{self._psnURL}41L6bd712.png")
e.set_image(url=f"attachment://{imgName}.png")
self._send_embed(e, file)
def send_death(self, location, image_path):
file = discord.File(image_path, filename="death.png")
e = discord.Embed(title=f"{self._config.general['name']} has died at {location}", color=Color.dark_red())
e.title=(f"{self._config.general['name']} died")
e.description=(f"Died at {location}")
e.set_thumbnail(url=f"{self._psnURL}33L5e3600.png")
e.set_image(url="attachment://death.png")
self._send_embed(e, file)
def send_chicken(self, location, image_path):
file = discord.File(image_path, filename="chicken.png")
e = discord.Embed(title=f"{self._config.general['name']} has chickened at {location}", color=Color.dark_grey())
e.title=(f"{self._config.general['name']} ran away")
e.description=(f"chickened at {location}")
e.set_thumbnail(url=f"{self._psnURL}39Ldf113b.png")
e.set_image(url="attachment://chicken.png")
self._send_embed(e, file)
def send_stash(self):
e = discord.Embed(title=f"{self._config.general['name']} has a full stash!", color=Color.dark_grey())
e.title=(f"{self._config.general['name']} has a full stash!")
e.description=(f"{self._config.general['name']} has to quit. \n They cannot store anymore items!")
e.set_thumbnail(url=f"{self._psnURL}35L63a9df.png")
self._send_embed(e)
def send_gold(self):
e = discord.Embed(title=f"{self._config.general['name']} is rich!", color=Color.dark_grey())
e.title=(f"{self._config.general['name']} is Rich!")
e.description=(f"{self._config.general['name']} can't store any more money!\n turning off gold pickup.")
e.set_thumbnail(url=f"{self._psnURL}6L341955.png")
self._send_embed(e)
def send_message(self, msg: str):
e = discord.Embed(title=f"Update:", description=f"```{msg}```", color=Color.dark_teal())
if not self._config.general['discord_status_condensed']:
e.set_thumbnail(url=f"{self._psnURL}36L4a4994.png")
self._send_embed(e)
def _send_embed(self, e, file = None):
e.set_footer(text=f'Botty v.{__version__} by Aeon')
e.timestamp=datetime.datetime.now(datetime.timezone.utc)
self._webhook.send(embed=e, file=file, username=self._config.general['name'])
def _get_Item_Color(self, item):
if "magic_" in item:
return Color.blue()
elif "set_" in item:
return Color.green()
elif "rune_" in item:
return Color.dark_gold()
elif "uniq_" in item or "rare" in item:
return Color.gold()
elif "gray_" in item:
return Color.darker_grey()
else:
return Color.blue()
src/api/generic_api.py
+59
View File
@@ -0,0 +1,59 @@
from config import Config
import numpy as np
import json
import requests
class GenericApi:
def __init__(self):
self._config = Config()
def send_item(self, item: str, image: np.ndarray, location: str):
msg = f"Found {item} at {location}"
self._send(msg)
def send_death(self, location: str, image_path: str = None):
msg = f"You have died at {location}"
self._send(msg)
def send_chicken(self, location: str, image_path: str = None):
msg = f"You have chickened at {location}"
self._send(msg)
def send_stash(self):
msg = f"All stash tabs and character are full of gold, turn of gold pickup"
self._send(msg)
def send_gold(self):
msg = f"All stash is full, quitting"
self._send(msg)
def send_message(self, msg: str):
self._send(msg)
def _send(self, msg: str):
if self._config.advanced_options['message_highlight']:
if " magic_" in msg:
msg = f"```ini\\n[ {msg} \\n```"
elif " set_" in msg:
msg = f"```diff\\n+ {msg} \\n```"
elif " rune_" in msg:
msg = f"```css\\n[ {msg} ]\\n```"
elif " uniq_" in msg or "rare" in msg:
# TODO: It is more gold than yellow, find a better yellow highlight
msg = f"```fix\\n- {msg} \\n```"
elif " gray_" in msg:
msg = f"```python\\n# {msg} \\n```"
else:
msg = f"```\\n{msg} \\n```"
url = self._config.general['custom_message_hook']
if not url:
return
headers = {}
if self._config.advanced_options['message_headers']:
headers = json.loads(self._config.advanced_options['message_headers'])
data = json.loads(self._config.advanced_options['message_body_template'].format(msg=msg), strict=False)
requests.post(url, headers=headers, json=data)
src/bot.py
+1 -1
View File
@@ -204,7 +204,7 @@ class Bot:
hot_ip = self._config.dclone["dclone_hotip"]
Logger.debug(f"Current Game IP: {cur_game_ip} and HOTIP: {hot_ip}")
if hot_ip == cur_game_ip:
self._messenger.send(msg=f"Dclone IP Found on IP: {cur_game_ip}")
self._messenger.send_message(f"Dclone IP Found on IP: {cur_game_ip}")
print("Press Enter")
input()
os._exit(1)
src/config.py
+1
View File
@@ -56,6 +56,7 @@ class Config:
"logg_lvl": self._select_val("general", "logg_lvl"),
"randomize_runs": bool(int(self._select_val("general", "randomize_runs"))),
"difficulty": self._select_val("general", "difficulty"),
"message_api_type": self._select_val("general", "message_api_type"),
"custom_message_hook": self._select_val("general", "custom_message_hook"),
"discord_status_count": False if not self._select_val("general", "discord_status_count") else int(self._select_val("general", "discord_status_count")),
"discord_status_condensed": bool(int(self._select_val("general", "discord_status_condensed"))),
src/death_manager.py
+6 -1
View File
@@ -18,6 +18,7 @@ class DeathManager:
self._do_monitor = False
self._loop_delay = 1.0
self._callback = None
self._last_death_screenshot = None
def get_loop_delay(self):
return self._loop_delay
@@ -42,9 +43,13 @@ class DeathManager:
mouse.click(button="left")
def handle_death_screen(self):
template_match = self._template_finder.search("YOU_HAVE_DIED", self._screen.grab(), threshold=0.9, roi=self._config.ui_roi["death"])
img = self._screen.grab()
template_match = self._template_finder.search("YOU_HAVE_DIED", img, threshold=0.9, roi=self._config.ui_roi["death"])
if template_match.valid:
Logger.warning("You have died!")
if self._config.general["info_screenshots"]:
self._last_death_screenshot = "./info_screenshots/info_debug_death_" + time.strftime("%Y%m%d_%H%M%S") + ".png"
cv2.imwrite(self._last_death_screenshot, img)
# first wait a bit to make sure health manager is done with its chicken stuff which obviously failed
if self._callback is not None:
self._callback()
src/game_controller.py
+5 -5
View File
@@ -56,9 +56,9 @@ class GameController:
if self._config.general["info_screenshots"]:
cv2.imwrite("./info_screenshots/info_max_game_length_reached_" + time.strftime("%Y%m%d_%H%M%S") + ".png", self.screen.grab())
elif self.death_manager.died():
self.game_stats.log_death()
self.game_stats.log_death(self.death_manager._last_death_screenshot)
elif self.health_manager.did_chicken():
self.game_stats.log_chicken()
self.game_stats.log_chicken(self.health_manager._last_chicken_screenshot)
self.bot.stop()
kill_thread(self.bot_thread)
# Try to recover from whatever situation we are and go back to hero selection
@@ -78,7 +78,7 @@ class GameController:
Logger.error(
f"{self._config.general['name']} could not recover from a max game length violation. Restarting the Game.")
if self._config.general["custom_message_hook"]:
messenger.send(msg=f"{self._config.general['name']}: got stuck and will now restart D2R")
messenger.send_message(f"{self._config.general['name']}: got stuck and will now restart D2R")
if restart_game(self._config.general["d2r_path"]):
self.game_stats.log_end_game(failed=max_game_length_reached)
if self.setup_screen():
@@ -87,8 +87,8 @@ class GameController:
self.game_recovery = GameRecovery(self.screen, self.death_manager)
return self.run_bot(True)
Logger.error(f"{self._config.general['name']} could not restart the game. Quitting.")
if self._config.general["custom_message_hook"]:
messenger.send(msg=f"{self._config.general['name']}: got stuck and will now quit")
messenger.send_message("Got stuck and could not restart the game. Quitting.")
os._exit(1)
def start(self):
src/game_stats.py
+31 -42
View File
@@ -1,3 +1,4 @@
import numpy as np
import time
import threading
import inspect
@@ -26,59 +27,48 @@ class GameStats:
self._failed_game_time = 0
self._location = None
self._location_stats = {}
self._location_stats["totals"] = { "items": 0, "deaths": 0, "chickens": 0, "merc_deaths": 0, "failed_runs": 0 }
self._stats_filename = f'stats_{time.strftime("%Y%m%d_%H%M%S")}.log'
def update_location(self, loc: str):
if self._location != loc:
self._location = str(loc)
self.populate_location_stat()
def get_location_msg(self):
if self._location is not None:
return f" at {self._location}"
else:
return ""
def _send_message_thread(self, msg: str):
if self._config.general["custom_message_hook"]:
send_message_thread = threading.Thread(
target=self._messenger.send,
kwargs={"msg": msg}
)
send_message_thread.daemon = True
send_message_thread.start()
def populate_location_stat(self):
if self._location not in self._location_stats:
self._location_stats[self._location] = { "items": [], "deaths": 0, "chickens": 0, "merc_deaths": 0, "failed_runs": 0 }
def log_item_keep(self, item_name: str, send_message: bool):
def log_item_keep(self, item_name: str, send_message: bool, img: np.ndarray):
filtered_items = ["_potion", "misc_gold"]
if self._location is not None and not any(substring in item_name for substring in filtered_items):
self._location_stats[self._location]["items"].append(item_name)
self._location_stats["totals"]["items"] += 1
if send_message:
msg = f"{self._config.general['name']}: Found {item_name}{self.get_location_msg()}"
self._send_message_thread(msg)
self._messenger.send_item(item_name, img, self._location)
def log_death(self):
def log_death(self, img: str):
self._death_counter += 1
if self._location is not None:
self._location_stats[self._location]["deaths"] += 1
msg = f"{self._config.general['name']}: You have died{self.get_location_msg()}"
self._send_message_thread(msg)
self._location_stats["totals"]["deaths"] += 1
self._messenger.send_death(self._location, img)
def log_chicken(self):
def log_chicken(self, img: str):
self._chicken_counter += 1
if self._location is not None:
self._location_stats[self._location]["chickens"] += 1
msg = f"{self._config.general['name']}: You have chickened{self.get_location_msg()}"
self._send_message_thread(msg)
self._location_stats["totals"]["chickens"] += 1
self._messenger.send_chicken(self._location, img)
def log_merc_death(self):
self._merc_death_counter += 1
if self._location is not None:
self._location_stats[self._location]["merc_deaths"] += 1
self._location_stats["totals"]["merc_deaths"] += 1
def log_start_game(self):
if self._game_counter > 0:
@@ -99,6 +89,7 @@ class GameStats:
self._runs_failed += 1
if self._location is not None:
self._location_stats[self._location]["failed_runs"] += 1
self._location_stats["totals"]["failed_runs"] += 1
self._failed_game_time += elapsed_time
Logger.warning(f"End failed game: Elpased time: {elapsed_time:.2f}s")
else:
@@ -134,30 +125,26 @@ class GameStats:
good_games_time = elapsed_time - self._failed_game_time
avg_length = good_games_time / float(good_games_count)
avg_length_str = hms(avg_length)
msg = inspect.cleandoc(f'''
Session length: {elapsed_time_str}
Games: {self._game_counter}
Avg Game Length: {avg_length_str}
''')
totals = { "items": 0, "chickens": 0, "deaths": 0, "merc_deaths": 0, "failed_runs": 0 }
msg = f'\nSession length: {elapsed_time_str}\nGames: {self._game_counter}\nAvg Game Length: {avg_length_str}'
table = BeautifulTable()
table.set_style(BeautifulTable.STYLE_BOX_ROUNDED)
for location in self._location_stats:
if location == "totals":
continue
stats = self._location_stats[location]
totals["items"] += len(stats["items"])
totals["chickens"] += stats["chickens"]
totals["deaths"] += stats["deaths"]
totals["merc_deaths"] += stats["merc_deaths"]
totals["failed_runs"] += stats["failed_runs"]
table.rows.append([location, len(stats["items"]), stats["chickens"], stats["deaths"], stats["merc_deaths"], stats["failed_runs"]])
table.rows.append([
"T" if self._config.general['discord_status_condensed'] else "Total",
totals["items"],
totals["chickens"],
totals["deaths"],
totals["merc_deaths"],
totals["failed_runs"]
self._location_stats["totals"]["items"],
self._location_stats["totals"]["chickens"],
self._location_stats["totals"]["deaths"],
self._location_stats["totals"]["merc_deaths"],
self._location_stats["totals"]["failed_runs"]
])
if self._config.general['discord_status_condensed']:
table.columns.header = ["Run", "I", "C", "D", "MD", "F"]
else:
@@ -167,13 +154,15 @@ class GameStats:
return msg
def _send_status_update(self):
msg = f"{self._config.general['name']}: Status Report\\n{self._create_msg()}\\nVersion: {__version__}"
self._send_message_thread(msg)
msg = f"{self._config.general['name']}: Status Report\n{self._create_msg()}\nVersion: {__version__}"
self._messenger.send_message(msg)
def _save_stats_to_file(self):
msg = self._create_msg()
msg += "\nItems:"
for location in self._location_stats:
if location == "totals":
continue
stats = self._location_stats[location]
msg += f"\n {location}:"
for item_name in stats["items"]:
src/health_manager.py
+3 -1
View File
@@ -29,6 +29,7 @@ class HealthManager:
self._last_merc_healh = time.time()
self._callback = None
self._pausing = True
self._last_chicken_screenshot = None
def stop_monitor(self):
self._do_monitor = False
@@ -91,7 +92,8 @@ class HealthManager:
self._callback()
self._callback = None
if self._config.general["info_screenshots"]:
cv2.imwrite("./info_screenshots/info_debug_chicken_" + time.strftime("%Y%m%d_%H%M%S") + ".png", img)
self._last_chicken_screenshot = "./info_screenshots/info_debug_chicken_" + time.strftime("%Y%m%d_%H%M%S") + ".png"
cv2.imwrite(self._last_chicken_screenshot, img)
# clean up key presses that might be pressed in the run_thread
keyboard.release(self._config.char["stand_still"])
wait(0.02, 0.05)
src/main.py
+1 -1
View File
@@ -59,7 +59,7 @@ def main():
os.system("mkdir stats")
if not os.path.exists("info_screenshots") and config.general["info_screenshots"]:
os.system("mkdir info_screenshots")
if not os.path.exists("loot_screenshots") and config.general["loot_screenshots"]:
if not os.path.exists("loot_screenshots") and (config.general["loot_screenshots"] or config.general["message_api_type"] == "discord"):
os.system("mkdir loot_screenshots")
print(f"============ Botty {__version__} [name: {config.general['name']}] ============")
src/messenger.py
+36 -32
View File
@@ -1,44 +1,48 @@
from dataclasses import dataclass
from config import Config
import json
import requests
import numpy as np
from api.generic_api import GenericApi
from api.discord_embeds import DiscordEmbeds
class Messenger:
def __init__(self):
self._config = Config()
if self._config.general["message_api_type"] == "generic_api":
self._message_api = GenericApi()
elif self._config.general["message_api_type"] == "discord":
self._message_api = DiscordEmbeds()
else:
self._message_api = None
def send(self, msg):
if self._config.advanced_options['message_highlight']:
if " magic_" in msg:
msg = f"```ini\\n[ {msg} \\n```"
elif " set_" in msg:
msg = f"```diff\\n+ {msg} \\n```"
elif " rune_" in msg:
msg = f"```css\\n[ {msg} ]\\n```"
elif " uniq_" in msg or "rare" in msg:
# TODO: It is more gold than yellow, find a better yellow highlight
msg = f"```fix\\n- {msg} \\n```"
elif " gray_" in msg:
msg = f"```python\\n# {msg} \\n```"
else:
msg = f"```\\n{msg} \\n```"
def send_item(self, item: str, image: np.ndarray, location: str):
self._message_api.send_item(item, image, location)
def send_death(self, location: str, image_path: str = None):
self._message_api.send_death(location, image_path)
def send_chicken(self, location: str, image_path: str = None):
self._message_api.send_chicken(location, image_path)
def send_stash(self):
self._message_api.send_stash()
self._send(msg=msg)
def _send(self, msg):
url = self._config.general['custom_message_hook']
if not url:
return
headers = {}
if self._config.advanced_options['message_headers']:
headers = json.loads(self._config.advanced_options['message_headers'])
data = json.loads(self._config.advanced_options['message_body_template'].format(msg=msg), strict=False)
requests.post(url, headers=headers, json=data)
def send_gold(self):
self._message_api.send_gold()
def send_message(self, msg: str):
self._message_api.send_message(msg)
if __name__ == "__main__":
messenger = Messenger()
messenger.send(msg=f" uniq_test")
item = "rune_test"
image = None
location = "Shenk"
# messenger.send_item(item, img, location)
# messenger.send_death(location, "./info_screenshots/info_debug_chicken_20211220_110621.png")
# messenger.send_chicken(location, "./info_screenshots/info_debug_chicken_20211220_110621.png")
messenger.send_stash()
messenger.send_gold()
messenger.send_message("This is a test message")
src/shop/anya.py
+6 -4
View File
@@ -129,7 +129,8 @@ class AnyaShopper:
)
if gg_gloves.valid:
mouse.click(button="right")
self._messenger.send(msg=f"{self._config.general['name']}: Bought awesome IAS/+3 gloves!")
self._messenger.send_message("Bought awesome IAS/+3 gloves!")
Logger.info("IAS/+3 gloves bought!")
self.gloves_bought += 1
time.sleep(1)
@@ -146,7 +147,7 @@ class AnyaShopper:
)
if g_gloves.valid:
mouse.click(button="right")
self._messenger.send(msg=f"{self._config.general['name']}: Bought some decent IAS/+2 gloves")
self._messenger.send_message("Bought some decent IAS/+2 gloves")
Logger.info("IAS/+2 gloves bought!")
self.gloves_bought += 1
time.sleep(1)
@@ -207,7 +208,8 @@ class AnyaShopper:
if trap_score > self.trap_claw_min_score and self.look_for_trap_claws is True:
# pick it up
mouse.click(button="right")
self._messenger.send(msg=f"{self._config.general['name']}: Bought some terrific trap Claws (score: {trap_score})")
self._messenger.send_message(f"Bought some terrific trap Claws (score: {trap_score})")
Logger.info(f"Trap Claws (score: {trap_score}) bought!")
self.claws_bought += 1
time.sleep(1)
@@ -215,7 +217,7 @@ class AnyaShopper:
if melee_score > self.melee_claw_min_score and self.look_for_melee_claws is True:
# pick it up
mouse.click(button="right")
self._messenger.send(msg=f"{self._config.general['name']}: Bought some mad melee Claws (score: {melee_score})")
self._messenger.send_message(f"Bought some mad melee Claws (score: {melee_score})")
Logger.info(f"Melee Claws (score: {melee_score}) bought!")
self.claws_bought += 1
time.sleep(1)
src/ui/ui_manager.py
+5 -6
View File
@@ -270,7 +270,7 @@ class UiManager():
exclude_props = self._config.items[x.name].exclude
if not (include_props or exclude_props):
Logger.debug(f"{x.name}: Stashing")
self._game_stats.log_item_keep(x.name, self._config.items[x.name].pickit_type == 2)
self._game_stats.log_item_keep(x.name, self._config.items[x.name].pickit_type == 2, img)
filtered_list.append(x)
continue
include = True
@@ -320,7 +320,7 @@ class UiManager():
break
if include and not exclude:
Logger.debug(f"{x.name}: Stashing. Required {include_logic_type}({include_props})={include}, exclude {exclude_logic_type}({exclude_props})={exclude}")
self._game_stats.log_item_keep(x.name, self._config.items[x.name].pickit_type == 2)
self._game_stats.log_item_keep(x.name, self._config.items[x.name].pickit_type == 2, img)
filtered_list.append(x)
return len(filtered_list) > 0
@@ -389,10 +389,9 @@ class UiManager():
self._config.items["misc_gold"].pickit_type = 0
item_finder.update_items_to_pick(self._config)
# inform user about it
msg = "All stash tabs and character are full of gold, turn of gold pickup"
Logger.info(msg)
Logger.info("All stash tabs and character are full of gold, turn of gold pickup")
if self._config.general["custom_message_hook"]:
self._messenger.send(msg=f"{self._config.general['name']}: {msg}")
self._messenger.send_gold()
else:
# move to next stash
wait(0.5, 0.6)
@@ -454,7 +453,7 @@ class UiManager():
if self._curr_stash["items"] > 3:
Logger.error("All stash is full, quitting")
if self._config.general["custom_message_hook"]:
self._messenger.send(msg=f"{self._config.general['name']}: all stash is full, quitting")
self._messenger.send_stash()
os._exit(1)
else:
# move to next stash
src/utils/dclone_ip.py
+1 -1
View File
@@ -49,6 +49,6 @@ if __name__ == "__main__":
if config.dclone["region_ips"] != "" and config.dclone["dclone_hotip"] != "":
print(f"Current Game IP: {get_d2r_game_ip()}")
print(f"Current Game Server: {get_d2r_game_server_region_by_ip(get_d2r_game_ip())}")
messenger.send(msg=f"Dclone IP Found on {get_d2r_game_server_region_by_ip(get_d2r_game_ip())} on IP: {get_d2r_game_ip()}")
messenger.send_message(f"Dclone IP Found on {get_d2r_game_server_region_by_ip(get_d2r_game_ip())} on IP: {get_d2r_game_ip()}")
else:
print(f"Please Enter the region ip and hot ip on config to use")