import os import sys import asyncio import threading import importlib.util import Utils import pathlib from typing import Dict, NamedTuple, Optional from NetUtils import ClientStatus from CommonClient import gui_enabled, logger, get_base_parser, ClientCommandProcessor, \ CommonContext, server_loop import settings # --------------------- Styx Scribe useful globals ----------------- global subsume styx_scribe_recieve_prefix = "Polycosmos to Client:" styx_scribe_send_prefix = "Client to Polycosmos:" # --------------------- Styx Scribe useful globals ----------------- # Here we implement methods for the client class HadesClientCommandProcessor(ClientCommandProcessor): def _cmd_resync(self): """Manually trigger a resync.""" #This is a really stupid solution, but it works so idk Utils.async_start(self.ctx.check_connection_and_send_items_and_request_starting_info("")) class HadesContext(CommonContext): # ----------------- Client start up and ending section starts -------------------------------- command_processor = HadesClientCommandProcessor game = "Hades" items_handling = 0b111 # full remote polycosmos_version = "0.15" is_connected : bool deathlink_pending : bool deathlink_enabled : bool creating_location_to_item_mapping : bool is_receiving_items_from_connect_package : bool location_name_to_id : dict def __init__(self, server_address: Optional[str] = None, password: Optional[str] = None): super(HadesContext, self).__init__(server_address, password) self.hades_slot_data = None self.is_connected = False self.deathlink_pending = False self.deathlink_enabled = False self.creating_location_to_item_mapping = False self.is_receiving_items_from_connect_package = False self.missing_locations_cache = [] self.checked_locations_cache = [] # Add hook to comunicate with StyxScribe subsume.AddHook(self.send_location_check_to_server, styx_scribe_recieve_prefix + "Locations updated:", "HadesClient") subsume.AddHook(self.on_run_completion, styx_scribe_recieve_prefix + "Hades defeated", "HadesClient") subsume.AddHook(self.check_connection_and_send_items_and_request_starting_info, styx_scribe_recieve_prefix + "Data requested", "HadesClient") # hook to send deathlink to other player when Zag dies subsume.AddHook(self.send_death, styx_scribe_recieve_prefix + "Zag died", "HadesClient") subsume.AddHook(self.send_location_hint_to_server, styx_scribe_recieve_prefix \ + "Locations hinted:", "HadesClient") async def server_auth(self, password_requested: bool = False) -> None: # This is called to autentificate with the server. if password_requested and not self.password: await super(HadesContext, self).server_auth(password_requested) await self.get_username() self.tags = set() await self.send_connect() async def connection_closed(self) -> None: # This is called when the connection is closed (duh!) # This will send the message always, but only process by Styx scribe if actually in game subsume.Send(styx_scribe_send_prefix + "Connection Error") self.is_connected = False self.is_receiving_items_from_connect_package = False await super(HadesContext, self).connection_closed() # Do not touch this @property def endpoints(self): if self.server: return [self.server] else: return [] async def shutdown(self): # What is called when the app gets shutdown subsume.close() await super(HadesContext, self).shutdown() # ----------------- Client start up and ending section end -------------------------------- # ----------------- Package Management section starts -------------------------------- def on_package(self, cmd: str, args: dict) -> None: # This is what is done when a package arrives. if cmd == "Connected": # What should be done in a connection package self.hades_slot_data = args["slot_data"] if not (self.hades_slot_data["version_check"] == self.polycosmos_version): stringError = "WORLD GENERATED WITH POLYCOSMOS " + self.hades_slot_data["version_check"] \ + " AND CLIENT USING POLYCOSMOS " + self.polycosmos_version + "\n" stringError += "THESE ARE NOT COMPATIBLE" raise Exception(stringError) self.location_name_to_id = self.get_location_name_to_id() if "death_link" in self.hades_slot_data and self.hades_slot_data["death_link"]: Utils.async_start(self.update_death_link(True)) self.deathlink_enabled = True self.is_connected = True self.is_receiving_items_from_connect_package = True if cmd == "RoomInfo": # What should be done when room info is sent. self.seed_name = args["seed_name"] if cmd == "RoomUpdate": if "checked_lodations" in args and len(args["checked_locations"]) > 0: subsume.Send(styx_scribe_send_prefix + "Locations collected:" + "-".join( (location) for location in args["checked_locations"] )) if cmd == "ReceivedItems": # We ignore sending the package to hades if just connected, # since the game is not ready for it (and will request it itself later) if self.is_receiving_items_from_connect_package: return self.send_items() if cmd == "LocationInfo": if self.creating_location_to_item_mapping: self.creating_location_to_item_mapping = False self.create_location_to_item_dictionary(args["locations"]) return super().on_package(cmd, args) if cmd == "Bounced": if "tags" in args: if "DeathLink" in args["tags"]: self.on_deathlink(args["data"]) def send_items(self) -> None: payload_message = ",".join(self.item_names.lookup_in_game(item.item) for item in self.items_received) subsume.Send(styx_scribe_send_prefix + "Items Updated:" + payload_message) async def send_location_check_to_server(self, message : str) -> None: if (self.location_name_to_id): await self.check_locations([self.location_name_to_id[message]]) async def check_connection_and_send_items_and_request_starting_info(self, message : str) -> None: if self.check_for_connection(): self.is_receiving_items_from_connect_package = False # send items that were already cached in connect self.send_items() self.request_location_to_item_dictionary() def store_settings_data(self) -> None: hades_settings_string = "" #codify in the string all heat settings hades_settings_string += str(self.hades_slot_data["heat_system"]) + "-" hades_settings_string += str(self.hades_slot_data["hard_labor_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["lasting_consequences_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["convenience_fee_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["jury_summons_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["extreme_measures_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["calisthenics_program_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["benefits_package_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["middle_management_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["underworld_customs_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["forced_overtime_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["heightened_security_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["routine_inspection_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["damage_control_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["approval_process_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["tight_deadline_pact_amount"]) + "-" hades_settings_string += str(self.hades_slot_data["personal_liability_pact_amount"]) + "-" #Codify in the string all the fillers values hades_settings_string += str(self.hades_slot_data["darkness_pack_value"]) + "-" hades_settings_string += str(self.hades_slot_data["keys_pack_value"]) + "-" hades_settings_string += str(self.hades_slot_data["gemstones_pack_value"]) + "-" hades_settings_string += str(self.hades_slot_data["diamonds_pack_value"]) + "-" hades_settings_string += str(self.hades_slot_data["titan_blood_pack_value"]) + "-" hades_settings_string += str(self.hades_slot_data["nectar_pack_value"]) + "-" hades_settings_string += str(self.hades_slot_data["ambrosia_pack_value"]) + "-" #Codify in the string all game settings hades_settings_string += str(self.hades_slot_data["location_system"]) + "-" hades_settings_string += str(self.hades_slot_data["reverse_order_em"]) + "-" hades_settings_string += str(self.hades_slot_data["keepsakesanity"]) + "-" hades_settings_string += str(self.hades_slot_data["weaponsanity"]) + "-" hades_settings_string += str(self.hades_slot_data["storesanity"]) + "-" hades_settings_string += str(self.hades_slot_data["initial_weapon"]) + "-" hades_settings_string += str(self.hades_slot_data["ignore_greece_deaths"]) + "-" hades_settings_string += str(self.hades_slot_data["fatesanity"]) + "-" hades_settings_string += str(self.hades_slot_data["hidden_aspectsanity"]) + "-" hades_settings_string += str(self.polycosmos_version) + "-" hades_settings_string += str(self.hades_slot_data["automatic_rooms_finish_on_hades_defeat"]) + "-" #Codify in the string all the finishing conditions hades_settings_string += str(self.hades_slot_data["hades_defeats_needed"]) + "-" hades_settings_string += str(self.hades_slot_data["weapons_clears_needed"]) + "-" hades_settings_string += str(self.hades_slot_data["keepsakes_needed"]) + "-" hades_settings_string += str(self.hades_slot_data["fates_needed"]) + "-" return hades_settings_string def request_location_to_item_dictionary(self) -> None: self.creating_location_to_item_mapping = True request = self.server_locations Utils.async_start(self.send_msgs([{"cmd": "LocationScouts", "locations": request, "create_as_hint": 0}])) def create_location_to_item_dictionary(self, itemsdict : Optional[dict]) -> None: locationItemMapping = "" for networkitem in itemsdict: location = self.parse_to_len_encode(self.location_names.lookup_in_slot(networkitem.location)) player_name = self.parse_to_len_encode(self.player_names[networkitem.player]) item_name = self.parse_to_len_encode(self.item_names.lookup_in_slot(networkitem.item, networkitem.player)) locationItemMapping += self.parse_to_len_encode(location+player_name+item_name) subsume.Send(styx_scribe_send_prefix + "Location to Item Map:" + locationItemMapping) subsume.Send(styx_scribe_send_prefix + "Data finished" + self.store_settings_data()) def parse_to_len_encode(self, inputstring: str) -> str: output = self.clear_invalid_char(inputstring) return str(len(output)) + "|" + output def obtain_array_from_len_encode(self, inputstring: str) -> list: result = [] if (inputstring == ""): return result run_index = 0 while run_index < len(inputstring): sep_index = 0 for i in range(run_index, len(inputstring)): char_index = inputstring[i] if (char_index == "|"): sep_index = i break if (run_index == sep_index): break lenmessage = int(inputstring[run_index : sep_index]) word = inputstring[sep_index + 1 : sep_index + lenmessage + 1] result.append(word) run_index = sep_index + lenmessage + 1 return result def clear_invalid_char(self, inputstring: str) -> str: newstr = inputstring.replace("{", "") newstr = newstr.replace("}", "") newstr = newstr.encode("ascii", "replace").decode(encoding="utf-8", errors="ignore") return newstr # ----------------- Package Management section ends -------------------------------- # ----------------- Hints from game section starts -------------------------------- async def send_location_hint_to_server(self, message : str) -> None: if self.hades_slot_data["store_give_hints"] == 0: return split_array = self.obtain_array_from_len_encode(message) request = [] for location in split_array: if len(location) > 0: request.append(self.location_name_to_id[location]) Utils.async_start(self.send_msgs([{"cmd": "LocationScouts", "locations": request, "create_as_hint": 2}])) # ----------------- Hints from game section ends ------------------------ # -------------deathlink section started -------------------------------- def on_deathlink(self, data: dict) -> None: # What should be done when a deathlink message is recieved if self.deathlink_pending: return self.deathlink_pending = True subsume.Send(styx_scribe_send_prefix + "Deathlink received") super().on_deathlink(data) Utils.async_start(self.wait_and_lower_deathlink_flag()) def send_death(self, death_text: str = "") -> None: # What should be done to send a death link # Avoid sending death if we died from a deathlink if self.deathlink_pending or not self.deathlink_enabled: return self.deathlink_pending = True Utils.async_start(super().send_death(death_text)) Utils.async_start(self.wait_and_lower_deathlink_flag()) async def wait_and_lower_deathlink_flag(self) -> None: await asyncio.sleep(3) self.deathlink_pending = False # -------------deathlink section ended # -------------game completion section starts # this is to detect game completion. Note that on futher updates this will need -------------------------------- # to be changed to adapt to new game completion conditions def on_run_completion(self, message : str) -> None: #parse message counters = message.split("-") #counters[0] is number of clears, counters[1] is number of different weapons with runs clears. hasEnoughRuns = self.hades_slot_data["hades_defeats_needed"] <= int(counters[0]) hasEnoughWeapons = self.hades_slot_data["weapons_clears_needed"] <= int(counters[1]) hasEnoughKeepsakes = self.hades_slot_data["keepsakes_needed"] <= int(counters[2]) hasEnoughFates = self.hades_slot_data["fates_needed"] <= int(counters[3]) if hasEnoughRuns and hasEnoughWeapons and hasEnoughKeepsakes and hasEnoughFates: Utils.async_start(self.send_msgs([{"cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL}])) self.finished_game = True # -------------game completion section ended -------------------------------- # ------------ game connection QoL handling def check_for_connection(self) -> bool: if not self.is_connected: subsume.Send(styx_scribe_send_prefix + "Connection Error") return False return True # ------------ Helper method to invert lookup table. Can erase if AP has its own internal one. def get_location_name_to_id(self): table = {} for locationid in self.server_locations: table[self.location_names.lookup_in_slot(locationid)] = locationid return table # ------------ gui section ------------------------------------------------ def run_gui(self) -> None: from kvui import GameManager class HadesManager(GameManager): # logging_pairs for any separate logging tabs base_title = "Archipelago Hades Client" self.ui = HadesManager(self) self.ui_task = Utils.async_start(self.ui.async_run(), name="UI") # ------------ Methods to start the client + Hades + StyxScribe ------------ def launch_hades(): subsume.Launch(True, None) def launch(): async def main(args): ctx = HadesContext(args.connect, args.password) ctx.server_task = Utils.async_start(server_loop(ctx), name="server loop") if gui_enabled: ctx.run_gui() ctx.run_cli() await ctx.exit_event.wait() ctx.server_address = None await ctx.shutdown() import colorama # --------------------- Styx Scribe initialization ----------------- styx_scribe_path = settings.get_settings()["hades_options"]["styx_scribe_path"] # Parsing of styxscribe path. This will try to find it on the same folder if it fails. last_slash = styx_scribe_path.rfind("/") if (last_slash == -1): print("Invalid path given to hades client for styxscribe. Cant parse") return filesbstr = styx_scribe_path[last_slash + 1 :] if (filesbstr != "StyxScribe.py"): print("Path given does not correspond to StyxScribe. Attempting to parse") styx_scribe_path = styx_scribe_path[:last_slash + 1] + "StyxScribe.py" hadespath = os.path.dirname(styx_scribe_path) if (not os.path.exists(hadespath)): print("Styx scribe not found at path.") spec = importlib.util.spec_from_file_location("StyxScribe", str(styx_scribe_path)) styx_scribe = importlib.util.module_from_spec(spec) sys.modules["StyxScribe"] = styx_scribe spec.loader.exec_module(styx_scribe) global subsume subsume = styx_scribe.StyxScribe("Hades") # hack to make it work without chdir subsume.proxy_purepaths = { None: hadespath / subsume.executable_cwd_purepath / styx_scribe.LUA_PROXY_STDIN, False: hadespath / subsume.executable_cwd_purepath / styx_scribe.LUA_PROXY_FALSE, True: hadespath / subsume.executable_cwd_purepath / styx_scribe.LUA_PROXY_TRUE } subsume.args[0] = os.path.normpath(os.path.join(hadespath, subsume.args[0])) subsume.executable_purepath = pathlib.PurePath(hadespath, subsume.executable_purepath) for i in range(len(subsume.plugins_paths)): subsume.plugins_paths[i] = pathlib.PurePath(hadespath, subsume.plugins_paths[i]) subsume.LoadPlugins() # --------------------- Styx Scribe initialization ----------------- thr = threading.Thread(target=launch_hades, args=(), kwargs={}) thr.start() parser = get_base_parser() args = parser.parse_args() colorama.init() asyncio.run(main(args)) colorama.deinit()