from typing import TYPE_CHECKING, Dict, Set from NetUtils import ClientStatus import worlds._bizhawk as bizhawk from worlds._bizhawk.client import BizHawkClient from .data import BASE_OFFSET, data from .options import Goal if TYPE_CHECKING: from worlds._bizhawk.context import BizHawkClientContext EXPECTED_ROM_NAME = "pokemon emerald version / AP 2" IS_CHAMPION_FLAG = data.constants["FLAG_IS_CHAMPION"] DEFEATED_STEVEN_FLAG = data.constants["TRAINER_FLAGS_START"] + data.constants["TRAINER_STEVEN"] DEFEATED_NORMAN_FLAG = data.constants["TRAINER_FLAGS_START"] + data.constants["TRAINER_NORMAN_1"] # These flags are communicated to the tracker as a bitfield using this order. # Modifying the order will cause undetectable autotracking issues. TRACKER_EVENT_FLAGS = [ "FLAG_DEFEATED_RUSTBORO_GYM", "FLAG_DEFEATED_DEWFORD_GYM", "FLAG_DEFEATED_MAUVILLE_GYM", "FLAG_DEFEATED_LAVARIDGE_GYM", "FLAG_DEFEATED_PETALBURG_GYM", "FLAG_DEFEATED_FORTREE_GYM", "FLAG_DEFEATED_MOSSDEEP_GYM", "FLAG_DEFEATED_SOOTOPOLIS_GYM", "FLAG_RECEIVED_POKENAV", # Talk to Mr. Stone "FLAG_DELIVERED_STEVEN_LETTER", "FLAG_DELIVERED_DEVON_GOODS", "FLAG_HIDE_ROUTE_119_TEAM_AQUA", # Clear Weather Institute "FLAG_MET_ARCHIE_METEOR_FALLS", # Magma steals meteorite "FLAG_GROUDON_AWAKENED_MAGMA_HIDEOUT", # Clear Magma Hideout "FLAG_MET_TEAM_AQUA_HARBOR", # Aqua steals submarine "FLAG_TEAM_AQUA_ESCAPED_IN_SUBMARINE", # Clear Aqua Hideout "FLAG_HIDE_MOSSDEEP_CITY_SPACE_CENTER_MAGMA_NOTE", # Clear Space Center "FLAG_KYOGRE_ESCAPED_SEAFLOOR_CAVERN", "FLAG_HIDE_SKY_PILLAR_TOP_RAYQUAZA", # Rayquaza departs for Sootopolis "FLAG_OMIT_DIVE_FROM_STEVEN_LETTER", # Steven gives Dive HM (clears seafloor cavern grunt) "FLAG_IS_CHAMPION", "FLAG_PURCHASED_HARBOR_MAIL" ] EVENT_FLAG_MAP = {data.constants[flag_name]: flag_name for flag_name in TRACKER_EVENT_FLAGS} KEY_LOCATION_FLAGS = [ "NPC_GIFT_RECEIVED_HM01", "NPC_GIFT_RECEIVED_HM02", "NPC_GIFT_RECEIVED_HM03", "NPC_GIFT_RECEIVED_HM04", "NPC_GIFT_RECEIVED_HM05", "NPC_GIFT_RECEIVED_HM06", "NPC_GIFT_RECEIVED_HM07", "NPC_GIFT_RECEIVED_HM08", "NPC_GIFT_RECEIVED_ACRO_BIKE", "NPC_GIFT_RECEIVED_WAILMER_PAIL", "NPC_GIFT_RECEIVED_DEVON_GOODS_RUSTURF_TUNNEL", "NPC_GIFT_RECEIVED_LETTER", "NPC_GIFT_RECEIVED_METEORITE", "NPC_GIFT_RECEIVED_GO_GOGGLES", "NPC_GIFT_GOT_BASEMENT_KEY_FROM_WATTSON", "NPC_GIFT_RECEIVED_ITEMFINDER", "NPC_GIFT_RECEIVED_DEVON_SCOPE", "NPC_GIFT_RECEIVED_MAGMA_EMBLEM", "NPC_GIFT_RECEIVED_POKEBLOCK_CASE", "NPC_GIFT_RECEIVED_SS_TICKET", "HIDDEN_ITEM_ABANDONED_SHIP_RM_2_KEY", "HIDDEN_ITEM_ABANDONED_SHIP_RM_1_KEY", "HIDDEN_ITEM_ABANDONED_SHIP_RM_4_KEY", "HIDDEN_ITEM_ABANDONED_SHIP_RM_6_KEY", "ITEM_ABANDONED_SHIP_HIDDEN_FLOOR_ROOM_4_SCANNER", "ITEM_ABANDONED_SHIP_CAPTAINS_OFFICE_STORAGE_KEY", "NPC_GIFT_RECEIVED_OLD_ROD", "NPC_GIFT_RECEIVED_GOOD_ROD", "NPC_GIFT_RECEIVED_SUPER_ROD", ] KEY_LOCATION_FLAG_MAP = {data.locations[location_name].flag: location_name for location_name in KEY_LOCATION_FLAGS} class PokemonEmeraldClient(BizHawkClient): game = "Pokemon Emerald" system = "GBA" patch_suffix = ".apemerald" local_checked_locations: Set[int] local_set_events: Dict[str, bool] local_found_key_items: Dict[str, bool] goal_flag: int def __init__(self) -> None: super().__init__() self.local_checked_locations = set() self.local_set_events = {} self.local_found_key_items = {} self.goal_flag = IS_CHAMPION_FLAG async def validate_rom(self, ctx: "BizHawkClientContext") -> bool: from CommonClient import logger try: # Check ROM name/patch version rom_name_bytes = ((await bizhawk.read(ctx.bizhawk_ctx, [(0x108, 32, "ROM")]))[0]) rom_name = bytes([byte for byte in rom_name_bytes if byte != 0]).decode("ascii") if not rom_name.startswith("pokemon emerald version"): return False if rom_name == "pokemon emerald version": logger.info("ERROR: You appear to be running an unpatched version of Pokemon Emerald. " "You need to generate a patch file and use it to create a patched ROM.") return False if rom_name != EXPECTED_ROM_NAME: logger.info("ERROR: The patch file used to create this ROM is not compatible with " "this client. Double check your client version against the version being " "used by the generator.") return False except UnicodeDecodeError: return False except bizhawk.RequestFailedError: return False # Should verify on the next pass ctx.game = self.game ctx.items_handling = 0b001 ctx.want_slot_data = True ctx.watcher_timeout = 0.125 return True async def set_auth(self, ctx: "BizHawkClientContext") -> None: slot_name_bytes = (await bizhawk.read(ctx.bizhawk_ctx, [(data.rom_addresses["gArchipelagoInfo"], 64, "ROM")]))[0] ctx.auth = bytes([byte for byte in slot_name_bytes if byte != 0]).decode("utf-8") async def game_watcher(self, ctx: "BizHawkClientContext") -> None: if ctx.slot_data is not None: if ctx.slot_data["goal"] == Goal.option_champion: self.goal_flag = IS_CHAMPION_FLAG elif ctx.slot_data["goal"] == Goal.option_steven: self.goal_flag = DEFEATED_STEVEN_FLAG elif ctx.slot_data["goal"] == Goal.option_norman: self.goal_flag = DEFEATED_NORMAN_FLAG try: # Checks that the player is in the overworld overworld_guard = (data.ram_addresses["gMain"] + 4, (data.ram_addresses["CB2_Overworld"] + 1).to_bytes(4, "little"), "System Bus") # Read save block address read_result = await bizhawk.guarded_read( ctx.bizhawk_ctx, [(data.ram_addresses["gSaveBlock1Ptr"], 4, "System Bus")], [overworld_guard] ) if read_result is None: # Not in overworld return # Checks that the save block hasn't moved save_block_address_guard = (data.ram_addresses["gSaveBlock1Ptr"], read_result[0], "System Bus") save_block_address = int.from_bytes(read_result[0], "little") # Handle giving the player items read_result = await bizhawk.guarded_read( ctx.bizhawk_ctx, [ (save_block_address + 0x3778, 2, "System Bus"), # Number of received items (data.ram_addresses["gArchipelagoReceivedItem"] + 4, 1, "System Bus") # Received item struct full? ], [overworld_guard, save_block_address_guard] ) if read_result is None: # Not in overworld, or save block moved return num_received_items = int.from_bytes(read_result[0], "little") received_item_is_empty = read_result[1][0] == 0 # If the game hasn't received all items yet and the received item struct doesn't contain an item, then # fill it with the next item if num_received_items < len(ctx.items_received) and received_item_is_empty: next_item = ctx.items_received[num_received_items] await bizhawk.write(ctx.bizhawk_ctx, [ (data.ram_addresses["gArchipelagoReceivedItem"] + 0, (next_item.item - BASE_OFFSET).to_bytes(2, "little"), "System Bus"), (data.ram_addresses["gArchipelagoReceivedItem"] + 2, (num_received_items + 1).to_bytes(2, "little"), "System Bus"), (data.ram_addresses["gArchipelagoReceivedItem"] + 4, [1], "System Bus"), # Mark struct full (data.ram_addresses["gArchipelagoReceivedItem"] + 5, [next_item.flags & 1], "System Bus"), ]) # Read flags in 2 chunks read_result = await bizhawk.guarded_read( ctx.bizhawk_ctx, [(save_block_address + 0x1450, 0x96, "System Bus")], # Flags [overworld_guard, save_block_address_guard] ) if read_result is None: # Not in overworld, or save block moved return flag_bytes = read_result[0] read_result = await bizhawk.guarded_read( ctx.bizhawk_ctx, [(save_block_address + 0x14E6, 0x96, "System Bus")], # Flags [overworld_guard, save_block_address_guard] ) if read_result is not None: flag_bytes += read_result[0] game_clear = False local_checked_locations = set() local_set_events = {flag_name: False for flag_name in TRACKER_EVENT_FLAGS} local_found_key_items = {location_name: False for location_name in KEY_LOCATION_FLAGS} # Check set flags for byte_i, byte in enumerate(flag_bytes): for i in range(8): if byte & (1 << i) != 0: flag_id = byte_i * 8 + i location_id = flag_id + BASE_OFFSET if location_id in ctx.server_locations: local_checked_locations.add(location_id) if flag_id == self.goal_flag: game_clear = True if flag_id in EVENT_FLAG_MAP: local_set_events[EVENT_FLAG_MAP[flag_id]] = True if flag_id in KEY_LOCATION_FLAG_MAP: local_found_key_items[KEY_LOCATION_FLAG_MAP[flag_id]] = True # Send locations if local_checked_locations != self.local_checked_locations: self.local_checked_locations = local_checked_locations if local_checked_locations is not None: await ctx.send_msgs([{ "cmd": "LocationChecks", "locations": list(local_checked_locations) }]) # Send game clear if not ctx.finished_game and game_clear: await ctx.send_msgs([{ "cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL }]) # Send tracker event flags if local_set_events != self.local_set_events and ctx.slot is not None: event_bitfield = 0 for i, flag_name in enumerate(TRACKER_EVENT_FLAGS): if local_set_events[flag_name]: event_bitfield |= 1 << i await ctx.send_msgs([{ "cmd": "Set", "key": f"pokemon_emerald_events_{ctx.team}_{ctx.slot}", "default": 0, "want_reply": False, "operations": [{"operation": "or", "value": event_bitfield}] }]) self.local_set_events = local_set_events if local_found_key_items != self.local_found_key_items: key_bitfield = 0 for i, location_name in enumerate(KEY_LOCATION_FLAGS): if local_found_key_items[location_name]: key_bitfield |= 1 << i await ctx.send_msgs([{ "cmd": "Set", "key": f"pokemon_emerald_keys_{ctx.team}_{ctx.slot}", "default": 0, "want_reply": False, "operations": [{"operation": "or", "value": key_bitfield}] }]) self.local_found_key_items = local_found_key_items except bizhawk.RequestFailedError: # Exit handler and return to main loop to reconnect pass