Archipelago/worlds/pokemon_emerald/client.py

278 lines
12 KiB
Python
Raw Normal View History

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