import ModuleUpdate
ModuleUpdate.update()
import os
import asyncio
import json
from pymem import pymem
from . import item_dictionary_table, exclusion_item_table, CheckDupingItems, all_locations, exclusion_table, SupportAbility_Table, ActionAbility_Table, all_weapon_slot
from .Names import ItemName
from .WorldLocations import *
from NetUtils import ClientStatus
from CommonClient import gui_enabled, logger, get_base_parser, CommonContext, server_loop
class KH2Context(CommonContext):
# command_processor: int = KH2CommandProcessor
game = "Kingdom Hearts 2"
items_handling = 0b111 # Indicates you get items sent from other worlds.
def __init__(self, server_address, password):
super(KH2Context, self).__init__(server_address, password)
self.goofy_ability_to_slot = dict()
self.donald_ability_to_slot = dict()
self.all_weapon_location_id = None
self.sora_ability_to_slot = dict()
self.kh2_seed_save = None
self.kh2_local_items = None
self.growthlevel = None
self.kh2connected = False
self.kh2_finished_game = False
self.serverconneced = False
self.item_name_to_data = {name: data for name, data, in item_dictionary_table.items()}
self.location_name_to_data = {name: data for name, data, in all_locations.items()}
self.kh2_loc_name_to_id = None
self.kh2_item_name_to_id = None
self.lookup_id_to_item = None
self.lookup_id_to_location = None
self.sora_ability_dict = {k: v.quantity for dic in [SupportAbility_Table, ActionAbility_Table] for k, v in
dic.items()}
self.location_name_to_worlddata = {name: data for name, data, in all_world_locations.items()}
self.sending = []
# list used to keep track of locations+items player has. Used for disoneccting
self.kh2_seed_save_cache = {
"itemIndex": -1,
# back of soras invo is 0x25E2. Growth should be moved there
# Character: [back of invo, front of invo]
"SoraInvo": [0x25D8, 0x2546],
"DonaldInvo": [0x26F4, 0x2658],
"GoofyInvo": [0x2808, 0x276C],
"AmountInvo": {
"Ability": {},
"Amount": {
"Bounty": 0,
},
"Growth": {
"High Jump": 0, "Quick Run": 0, "Dodge Roll": 0,
"Aerial Dodge": 0, "Glide": 0
},
"Bitmask": [],
"Weapon": {"Sora": [], "Donald": [], "Goofy": []},
"Equipment": [],
"Magic": {
"Fire Element": 0,
"Blizzard Element": 0,
"Thunder Element": 0,
"Cure Element": 0,
"Magnet Element": 0,
"Reflect Element": 0
},
"StatIncrease": {
ItemName.MaxHPUp: 0,
ItemName.MaxMPUp: 0,
ItemName.DriveGaugeUp: 0,
ItemName.ArmorSlotUp: 0,
ItemName.AccessorySlotUp: 0,
ItemName.ItemSlotUp: 0,
},
},
}
self.kh2seedname = None
self.kh2slotdata = None
self.itemamount = {}
if "localappdata" in os.environ:
self.game_communication_path = os.path.expandvars(r"%localappdata%\KH2AP")
self.hitlist_bounties = 0
# hooked object
self.kh2 = None
self.final_xemnas = False
self.worldid_to_locations = {
# 1: {}, # world of darkness (story cutscenes)
2: TT_Checks,
# 3: {}, # destiny island doesn't have checks
4: HB_Checks,
5: BC_Checks,
6: Oc_Checks,
7: AG_Checks,
8: LoD_Checks,
9: HundredAcreChecks,
10: PL_Checks,
11: Atlantica_Checks,
12: DC_Checks,
13: TR_Checks,
14: HT_Checks,
15: HB_Checks, # world map, but you only go to the world map while on the way to goa so checking hb
16: PR_Checks,
17: SP_Checks,
18: TWTNW_Checks,
# 255: {}, # starting screen
}
# 0x2A09C00+0x40 is the sve anchor. +1 is the last saved room
# self.sveroom = 0x2A09C00 + 0x41
# 0 not in battle 1 in yellow battle 2 red battle #short
# self.inBattle = 0x2A0EAC4 + 0x40
# self.onDeath = 0xAB9078
# PC Address anchors
self.Now = 0x0714DB8
self.Save = 0x09A70B0
# self.Sys3 = 0x2A59DF0
# self.Bt10 = 0x2A74880
# self.BtlEnd = 0x2A0D3E0
self.Slot1 = 0x2A20C98
self.chest_set = set(exclusion_table["Chests"])
self.keyblade_set = set(CheckDupingItems["Weapons"]["Keyblades"])
self.staff_set = set(CheckDupingItems["Weapons"]["Staffs"])
self.shield_set = set(CheckDupingItems["Weapons"]["Shields"])
self.all_weapons = self.keyblade_set.union(self.staff_set).union(self.shield_set)
self.equipment_categories = CheckDupingItems["Equipment"]
self.armor_set = set(self.equipment_categories["Armor"])
self.accessories_set = set(self.equipment_categories["Accessories"])
self.all_equipment = self.armor_set.union(self.accessories_set)
self.Equipment_Anchor_Dict = {
"Armor": [0x2504, 0x2506, 0x2508, 0x250A],
"Accessories": [0x2514, 0x2516, 0x2518, 0x251A]
}
self.AbilityQuantityDict = {}
self.ability_categories = CheckDupingItems["Abilities"]
self.sora_ability_set = set(self.ability_categories["Sora"])
self.donald_ability_set = set(self.ability_categories["Donald"])
self.goofy_ability_set = set(self.ability_categories["Goofy"])
self.all_abilities = self.sora_ability_set.union(self.donald_ability_set).union(self.goofy_ability_set)
self.stat_increase_set = set(CheckDupingItems["Stat Increases"])
self.AbilityQuantityDict = {item: self.item_name_to_data[item].quantity for item in self.all_abilities}
# Growth:[level 1,level 4,slot]
self.growth_values_dict = {
"High Jump": [0x05E, 0x061, 0x25DA],
"Quick Run": [0x62, 0x65, 0x25DC],
"Dodge Roll": [0x234, 0x237, 0x25DE],
"Aerial Dodge": [0x66, 0x069, 0x25E0],
"Glide": [0x6A, 0x6D, 0x25E2]
}
self.ability_code_list = None
self.master_growth = {"High Jump", "Quick Run", "Dodge Roll", "Aerial Dodge", "Glide"}
self.base_hp = 20
self.base_mp = 100
self.base_drive = 5
self.base_accessory_slots = 1
self.base_armor_slots = 1
self.base_item_slots = 3
self.front_ability_slots = [0x2546, 0x2658, 0x276C, 0x2548, 0x254A, 0x254C, 0x265A, 0x265C, 0x265E, 0x276E, 0x2770, 0x2772]
async def server_auth(self, password_requested: bool = False):
if password_requested and not self.password:
await super(KH2Context, self).server_auth(password_requested)
await self.get_username()
await self.send_connect()
async def connection_closed(self):
self.kh2connected = False
self.serverconneced = False
if self.kh2seedname is not None and self.auth is not None:
with open(os.path.join(self.game_communication_path, f"kh2save2{self.kh2seedname}{self.auth}.json"),
'w') as f:
f.write(json.dumps(self.kh2_seed_save, indent=4))
await super(KH2Context, self).connection_closed()
async def disconnect(self, allow_autoreconnect: bool = False):
self.kh2connected = False
self.serverconneced = False
if self.kh2seedname not in {None} and self.auth not in {None}:
with open(os.path.join(self.game_communication_path, f"kh2save2{self.kh2seedname}{self.auth}.json"),
'w') as f:
f.write(json.dumps(self.kh2_seed_save, indent=4))
await super(KH2Context, self).disconnect()
@property
def endpoints(self):
if self.server:
return [self.server]
else:
return []
async def shutdown(self):
if self.kh2seedname not in {None} and self.auth not in {None}:
with open(os.path.join(self.game_communication_path, f"kh2save2{self.kh2seedname}{self.auth}.json"),
'w') as f:
f.write(json.dumps(self.kh2_seed_save, indent=4))
await super(KH2Context, self).shutdown()
def kh2_read_short(self, address):
return self.kh2.read_short(self.kh2.base_address + address)
def kh2_write_short(self, address, value):
return self.kh2.write_short(self.kh2.base_address + address, value)
def kh2_write_byte(self, address, value):
return self.kh2.write_bytes(self.kh2.base_address + address, value.to_bytes(1, 'big'), 1)
def kh2_read_byte(self, address):
return int.from_bytes(self.kh2.read_bytes(self.kh2.base_address + address, 1), "big")
def kh2_read_int(self, address):
return self.kh2.read_int(self.kh2.base_address + address)
def kh2_write_int(self, address, value):
self.kh2.write_int(self.kh2.base_address + address, value)
def on_package(self, cmd: str, args: dict):
if cmd in {"RoomInfo"}:
self.kh2seedname = args['seed_name']
if not os.path.exists(self.game_communication_path):
os.makedirs(self.game_communication_path)
if not os.path.exists(self.game_communication_path + f"\kh2save2{self.kh2seedname}{self.auth}.json"):
self.kh2_seed_save = {
"Levels": {
"SoraLevel": 0,
"ValorLevel": 0,
"WisdomLevel": 0,
"LimitLevel": 0,
"MasterLevel": 0,
"FinalLevel": 0,
"SummonLevel": 0,
},
"SoldEquipment": [],
}
with open(os.path.join(self.game_communication_path, f"kh2save2{self.kh2seedname}{self.auth}.json"),
'wt') as f:
pass
# self.locations_checked = set()
elif os.path.exists(self.game_communication_path + f"\kh2save2{self.kh2seedname}{self.auth}.json"):
with open(self.game_communication_path + f"\kh2save2{self.kh2seedname}{self.auth}.json", 'r') as f:
self.kh2_seed_save = json.load(f)
if self.kh2_seed_save is None:
self.kh2_seed_save = {
"Levels": {
"SoraLevel": 0,
"ValorLevel": 0,
"WisdomLevel": 0,
"LimitLevel": 0,
"MasterLevel": 0,
"FinalLevel": 0,
"SummonLevel": 0,
},
"SoldEquipment": [],
}
# self.locations_checked = set(self.kh2_seed_save_cache["LocationsChecked"])
# self.serverconneced = True
if cmd in {"Connected"}:
asyncio.create_task(self.send_msgs([{"cmd": "GetDataPackage", "games": ["Kingdom Hearts 2"]}]))
self.kh2slotdata = args['slot_data']
# self.kh2_local_items = {int(location): item for location, item in self.kh2slotdata["LocalItems"].items()}
self.locations_checked = set(args["checked_locations"])
if cmd in {"ReceivedItems"}:
# 0x2546
# 0x2658
# 0x276A
start_index = args["index"]
if start_index == 0:
self.kh2_seed_save_cache = {
"itemIndex": -1,
# back of soras invo is 0x25E2. Growth should be moved there
# Character: [back of invo, front of invo]
"SoraInvo": [0x25D8, 0x2546],
"DonaldInvo": [0x26F4, 0x2658],
"GoofyInvo": [0x2808, 0x276C],
"AmountInvo": {
"Ability": {},
"Amount": {
"Bounty": 0,
},
"Growth": {
"High Jump": 0, "Quick Run": 0, "Dodge Roll": 0,
"Aerial Dodge": 0, "Glide": 0
},
"Bitmask": [],
"Weapon": {"Sora": [], "Donald": [], "Goofy": []},
"Equipment": [],
"Magic": {
"Fire Element": 0,
"Blizzard Element": 0,
"Thunder Element": 0,
"Cure Element": 0,
"Magnet Element": 0,
"Reflect Element": 0
},
"StatIncrease": {
ItemName.MaxHPUp: 0,
ItemName.MaxMPUp: 0,
ItemName.DriveGaugeUp: 0,
ItemName.ArmorSlotUp: 0,
ItemName.AccessorySlotUp: 0,
ItemName.ItemSlotUp: 0,
},
},
}
if start_index > self.kh2_seed_save_cache["itemIndex"] and self.serverconneced:
self.kh2_seed_save_cache["itemIndex"] = start_index
for item in args['items']:
asyncio.create_task(self.give_item(item.item, item.location))
if cmd in {"RoomUpdate"}:
if "checked_locations" in args:
new_locations = set(args["checked_locations"])
self.locations_checked |= new_locations
if cmd in {"DataPackage"}:
self.kh2_loc_name_to_id = args["data"]["games"]["Kingdom Hearts 2"]["location_name_to_id"]
self.lookup_id_to_location = {v: k for k, v in self.kh2_loc_name_to_id.items()}
self.kh2_item_name_to_id = args["data"]["games"]["Kingdom Hearts 2"]["item_name_to_id"]
self.lookup_id_to_item = {v: k for k, v in self.kh2_item_name_to_id.items()}
self.ability_code_list = [self.kh2_item_name_to_id[item] for item in exclusion_item_table["Ability"]]
if "keyblade_abilities" in self.kh2slotdata.keys():
sora_ability_dict = self.kh2slotdata["KeybladeAbilities"]
# sora ability to slot
# itemid:[slots that are available for that item]
for k, v in sora_ability_dict.items():
if v >= 1:
if k not in self.sora_ability_to_slot.keys():
self.sora_ability_to_slot[k] = []
for _ in range(sora_ability_dict[k]):
self.sora_ability_to_slot[k].append(self.kh2_seed_save_cache["SoraInvo"][0])
self.kh2_seed_save_cache["SoraInvo"][0] -= 2
donald_ability_dict = self.kh2slotdata["StaffAbilities"]
for k, v in donald_ability_dict.items():
if v >= 1:
if k not in self.donald_ability_to_slot.keys():
self.donald_ability_to_slot[k] = []
for _ in range(donald_ability_dict[k]):
self.donald_ability_to_slot[k].append(self.kh2_seed_save_cache["DonaldInvo"][0])
self.kh2_seed_save_cache["DonaldInvo"][0] -= 2
goofy_ability_dict = self.kh2slotdata["ShieldAbilities"]
for k, v in goofy_ability_dict.items():
if v >= 1:
if k not in self.goofy_ability_to_slot.keys():
self.goofy_ability_to_slot[k] = []
for _ in range(goofy_ability_dict[k]):
self.goofy_ability_to_slot[k].append(self.kh2_seed_save_cache["GoofyInvo"][0])
self.kh2_seed_save_cache["GoofyInvo"][0] -= 2
all_weapon_location_id = []
for weapon_location in all_weapon_slot:
all_weapon_location_id.append(self.kh2_loc_name_to_id[weapon_location])
self.all_weapon_location_id = set(all_weapon_location_id)
try:
self.kh2 = pymem.Pymem(process_name="KINGDOM HEARTS II FINAL MIX")
logger.info("You are now auto-tracking")
self.kh2connected = True
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info("Game is not open.")
self.serverconneced = True
asyncio.create_task(self.send_msgs([{'cmd': 'Sync'}]))
async def checkWorldLocations(self):
try:
currentworldint = self.kh2_read_byte(self.Now)
await self.send_msgs([{
"cmd": "Set", "key": "Slot: " + str(self.slot) + " :CurrentWorld",
"default": 0, "want_reply": True, "operations": [{
"operation": "replace",
"value": currentworldint
}]
}])
if currentworldint in self.worldid_to_locations:
curworldid = self.worldid_to_locations[currentworldint]
for location, data in curworldid.items():
if location in self.kh2_loc_name_to_id.keys():
locationId = self.kh2_loc_name_to_id[location]
if locationId not in self.locations_checked \
and self.kh2_read_byte(self.Save + data.addrObtained) & 0x1 << data.bitIndex > 0:
self.sending = self.sending + [(int(locationId))]
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info(e)
logger.info("line 425")
async def checkLevels(self):
try:
for location, data in SoraLevels.items():
currentLevel = self.kh2_read_byte(self.Save + 0x24FF)
locationId = self.kh2_loc_name_to_id[location]
if locationId not in self.locations_checked \
and currentLevel >= data.bitIndex:
if self.kh2_seed_save["Levels"]["SoraLevel"] < currentLevel:
self.kh2_seed_save["Levels"]["SoraLevel"] = currentLevel
self.sending = self.sending + [(int(locationId))]
formDict = {
0: ["ValorLevel", ValorLevels], 1: ["WisdomLevel", WisdomLevels], 2: ["LimitLevel", LimitLevels],
3: ["MasterLevel", MasterLevels], 4: ["FinalLevel", FinalLevels], 5: ["SummonLevel", SummonLevels]
}
# TODO: remove formDict[i][0] in self.kh2_seed_save_cache["Levels"].keys() after 4.3
for i in range(6):
for location, data in formDict[i][1].items():
formlevel = self.kh2_read_byte(self.Save + data.addrObtained)
if location in self.kh2_loc_name_to_id.keys():
# if current form level is above other form level
locationId = self.kh2_loc_name_to_id[location]
if locationId not in self.locations_checked \
and formlevel >= data.bitIndex:
if formlevel > self.kh2_seed_save["Levels"][formDict[i][0]]:
self.kh2_seed_save["Levels"][formDict[i][0]] = formlevel
self.sending = self.sending + [(int(locationId))]
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info(e)
logger.info("line 456")
async def checkSlots(self):
try:
for location, data in weaponSlots.items():
locationId = self.kh2_loc_name_to_id[location]
if locationId not in self.locations_checked:
if self.kh2_read_byte(self.Save + data.addrObtained) > 0:
self.sending = self.sending + [(int(locationId))]
for location, data in formSlots.items():
locationId = self.kh2_loc_name_to_id[location]
if locationId not in self.locations_checked and self.kh2_read_byte(self.Save + 0x06B2) == 0:
if self.kh2_read_byte(self.Save + data.addrObtained) & 0x1 << data.bitIndex > 0:
self.sending = self.sending + [(int(locationId))]
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info(e)
logger.info("line 475")
async def verifyChests(self):
try:
for location in self.locations_checked:
locationName = self.lookup_id_to_location[location]
if locationName in self.chest_set:
if locationName in self.location_name_to_worlddata.keys():
locationData = self.location_name_to_worlddata[locationName]
if self.kh2_read_byte(self.Save + locationData.addrObtained) & 0x1 << locationData.bitIndex == 0:
roomData = self.kh2_read_byte(self.Save + locationData.addrObtained)
self.kh2_write_byte(self.Save + locationData.addrObtained, roomData | 0x01 << locationData.bitIndex)
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info(e)
logger.info("line 491")
async def verifyLevel(self):
for leveltype, anchor in {
"SoraLevel": 0x24FF,
"ValorLevel": 0x32F6,
"WisdomLevel": 0x332E,
"LimitLevel": 0x3366,
"MasterLevel": 0x339E,
"FinalLevel": 0x33D6
}.items():
if self.kh2_read_byte(self.Save + anchor) < self.kh2_seed_save["Levels"][leveltype]:
self.kh2_write_byte(self.Save + anchor, self.kh2_seed_save["Levels"][leveltype])
async def give_item(self, item, location):
try:
# todo: ripout all the itemtype stuff and just have one dictionary. the only thing that needs to be tracked from the server/local is abilites
itemname = self.lookup_id_to_item[item]
itemdata = self.item_name_to_data[itemname]
# itemcode = self.kh2_item_name_to_id[itemname]
if itemdata.ability:
if location in self.all_weapon_location_id:
return
if itemname in {"High Jump", "Quick Run", "Dodge Roll", "Aerial Dodge", "Glide"}:
self.kh2_seed_save_cache["AmountInvo"]["Growth"][itemname] += 1
return
if itemname not in self.kh2_seed_save_cache["AmountInvo"]["Ability"]:
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname] = []
# appending the slot that the ability should be in
# for non beta. remove after 4.3
if "PoptrackerVersion" in self.kh2slotdata:
if self.kh2slotdata["PoptrackerVersionCheck"] < 4.3:
if (itemname in self.sora_ability_set
and len(self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname]) < self.item_name_to_data[itemname].quantity) \
and self.kh2_seed_save_cache["SoraInvo"][1] > 0x254C:
ability_slot = self.kh2_seed_save_cache["SoraInvo"][1]
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname].append(ability_slot)
self.kh2_seed_save_cache["SoraInvo"][1] -= 2
elif itemname in self.donald_ability_set:
ability_slot = self.kh2_seed_save_cache["DonaldInvo"][1]
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname].append(ability_slot)
self.kh2_seed_save_cache["DonaldInvo"][1] -= 2
else:
ability_slot = self.kh2_seed_save_cache["GoofyInvo"][1]
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname].append(ability_slot)
self.kh2_seed_save_cache["GoofyInvo"][1] -= 2
if ability_slot in self.front_ability_slots:
self.front_ability_slots.remove(ability_slot)
elif len(self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname]) < \
self.AbilityQuantityDict[itemname]:
if itemname in self.sora_ability_set:
ability_slot = self.kh2_seed_save_cache["SoraInvo"][0]
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname].append(ability_slot)
self.kh2_seed_save_cache["SoraInvo"][0] -= 2
elif itemname in self.donald_ability_set:
ability_slot = self.kh2_seed_save_cache["DonaldInvo"][0]
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname].append(ability_slot)
self.kh2_seed_save_cache["DonaldInvo"][0] -= 2
else:
ability_slot = self.kh2_seed_save_cache["GoofyInvo"][0]
self.kh2_seed_save_cache["AmountInvo"]["Ability"][itemname].append(ability_slot)
self.kh2_seed_save_cache["GoofyInvo"][0] -= 2
if ability_slot in self.front_ability_slots:
self.front_ability_slots.remove(ability_slot)
elif itemdata.memaddr in {0x36C4, 0x36C5, 0x36C6, 0x36C0, 0x36CA}:
# if memaddr is in a bitmask location in memory
if itemname not in self.kh2_seed_save_cache["AmountInvo"]["Bitmask"]:
self.kh2_seed_save_cache["AmountInvo"]["Bitmask"].append(itemname)
elif itemdata.memaddr in {0x3594, 0x3595, 0x3596, 0x3597, 0x35CF, 0x35D0}:
# if memaddr is in magic addresses
self.kh2_seed_save_cache["AmountInvo"]["Magic"][itemname] += 1
elif itemname in self.all_equipment:
self.kh2_seed_save_cache["AmountInvo"]["Equipment"].append(itemname)
elif itemname in self.all_weapons:
if itemname in self.keyblade_set:
self.kh2_seed_save_cache["AmountInvo"]["Weapon"]["Sora"].append(itemname)
elif itemname in self.staff_set:
self.kh2_seed_save_cache["AmountInvo"]["Weapon"]["Donald"].append(itemname)
else:
self.kh2_seed_save_cache["AmountInvo"]["Weapon"]["Goofy"].append(itemname)
elif itemname in self.stat_increase_set:
self.kh2_seed_save_cache["AmountInvo"]["StatIncrease"][itemname] += 1
else:
if itemname in self.kh2_seed_save_cache["AmountInvo"]["Amount"]:
self.kh2_seed_save_cache["AmountInvo"]["Amount"][itemname] += 1
else:
self.kh2_seed_save_cache["AmountInvo"]["Amount"][itemname] = 1
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info(e)
logger.info("line 582")
def run_gui(self):
"""Import kivy UI system and start running it as self.ui_task."""
from kvui import GameManager
class KH2Manager(GameManager):
logging_pairs = [
("Client", "Archipelago")
]
base_title = "Archipelago KH2 Client"
self.ui = KH2Manager(self)
self.ui_task = asyncio.create_task(self.ui.async_run(), name="UI")
async def IsInShop(self, sellable):
# journal = 0x741230 shop = 0x741320
# if journal=-1 and shop = 5 then in shop
# if journal !=-1 and shop = 10 then journal
journal = self.kh2_read_short(0x741230)
shop = self.kh2_read_short(0x741320)
if (journal == -1 and shop == 5) or (journal != -1 and shop == 10):
# print("your in the shop")
sellable_dict = {}
for itemName in sellable:
itemdata = self.item_name_to_data[itemName]
amount = self.kh2_read_byte(self.Save + itemdata.memaddr)
sellable_dict[itemName] = amount
while (journal == -1 and shop == 5) or (journal != -1 and shop == 10):
journal = self.kh2_read_short(0x741230)
shop = self.kh2_read_short(0x741320)
await asyncio.sleep(0.5)
for item, amount in sellable_dict.items():
itemdata = self.item_name_to_data[item]
afterShop = self.kh2_read_byte(self.Save + itemdata.memaddr)
if afterShop < amount:
self.kh2_seed_save["SoldEquipment"].append(item)
async def verifyItems(self):
try:
master_amount = set(self.kh2_seed_save_cache["AmountInvo"]["Amount"].keys())
master_ability = set(self.kh2_seed_save_cache["AmountInvo"]["Ability"].keys())
master_bitmask = set(self.kh2_seed_save_cache["AmountInvo"]["Bitmask"])
master_keyblade = set(self.kh2_seed_save_cache["AmountInvo"]["Weapon"]["Sora"])
master_staff = set(self.kh2_seed_save_cache["AmountInvo"]["Weapon"]["Donald"])
master_shield = set(self.kh2_seed_save_cache["AmountInvo"]["Weapon"]["Goofy"])
master_equipment = set(self.kh2_seed_save_cache["AmountInvo"]["Equipment"])
master_magic = set(self.kh2_seed_save_cache["AmountInvo"]["Magic"].keys())
master_stat = set(self.kh2_seed_save_cache["AmountInvo"]["StatIncrease"].keys())
master_sell = master_equipment | master_staff | master_shield
await asyncio.create_task(self.IsInShop(master_sell))
# print(self.kh2_seed_save_cache["AmountInvo"]["Ability"])
for item_name in master_amount:
item_data = self.item_name_to_data[item_name]
amount_of_items = 0
amount_of_items += self.kh2_seed_save_cache["AmountInvo"]["Amount"][item_name]
if item_name == "Torn Page":
# Torn Pages are handled differently because they can be consumed.
# Will check the progression in 100 acre and - the amount of visits
# amountofitems-amount of visits done
for location, data in tornPageLocks.items():
if self.kh2_read_byte(self.Save + data.addrObtained) & 0x1 << data.bitIndex > 0:
amount_of_items -= 1
if self.kh2_read_byte(self.Save + item_data.memaddr) != amount_of_items and amount_of_items >= 0:
self.kh2_write_byte(self.Save + item_data.memaddr, amount_of_items)
for item_name in master_keyblade:
item_data = self.item_name_to_data[item_name]
# if the inventory slot for that keyblade is less than the amount they should have,
# and they are not in stt
if self.kh2_read_byte(self.Save + item_data.memaddr) != 1 and self.kh2_read_byte(self.Save + 0x1CFF) != 13:
# Checking form anchors for the keyblade to remove extra keyblades
if self.kh2_read_short(self.Save + 0x24F0) == item_data.kh2id \
or self.kh2_read_short(self.Save + 0x32F4) == item_data.kh2id \
or self.kh2_read_short(self.Save + 0x339C) == item_data.kh2id \
or self.kh2_read_short(self.Save + 0x33D4) == item_data.kh2id:
self.kh2_write_byte(self.Save + item_data.memaddr, 0)
else:
self.kh2_write_byte(self.Save + item_data.memaddr, 1)
for item_name in master_staff:
item_data = self.item_name_to_data[item_name]
if self.kh2_read_byte(self.Save + item_data.memaddr) != 1 \
and self.kh2_read_short(self.Save + 0x2604) != item_data.kh2id \
and item_name not in self.kh2_seed_save["SoldEquipment"]:
self.kh2_write_byte(self.Save + item_data.memaddr, 1)
for item_name in master_shield:
item_data = self.item_name_to_data[item_name]
if self.kh2_read_byte(self.Save + item_data.memaddr) != 1 \
and self.kh2_read_short(self.Save + 0x2718) != item_data.kh2id \
and item_name not in self.kh2_seed_save["SoldEquipment"]:
self.kh2_write_byte(self.Save + item_data.memaddr, 1)
for item_name in master_ability:
item_data = self.item_name_to_data[item_name]
ability_slot = []
ability_slot += self.kh2_seed_save_cache["AmountInvo"]["Ability"][item_name]
for slot in ability_slot:
current = self.kh2_read_short(self.Save + slot)
ability = current & 0x0FFF
if ability | 0x8000 != (0x8000 + item_data.memaddr):
if current - 0x8000 > 0:
self.kh2_write_short(self.Save + slot, 0x8000 + item_data.memaddr)
else:
self.kh2_write_short(self.Save + slot, item_data.memaddr)
# removes the duped ability if client gave faster than the game.
for ability in self.front_ability_slots:
if self.kh2_read_short(self.Save + ability) != 0:
print(f"removed {self.Save + ability} from {ability}")
self.kh2_write_short(self.Save + ability, 0)
# remove the dummy level 1 growths if they are in these invo slots.
for inventorySlot in {0x25CE, 0x25D0, 0x25D2, 0x25D4, 0x25D6, 0x25D8}:
current = self.kh2_read_short(self.Save + inventorySlot)
ability = current & 0x0FFF
if 0x05E <= ability <= 0x06D:
self.kh2_write_short(self.Save + inventorySlot, 0)
for item_name in self.master_growth:
growthLevel = self.kh2_seed_save_cache["AmountInvo"]["Growth"][item_name]
if growthLevel > 0:
slot = self.growth_values_dict[item_name][2]
min_growth = self.growth_values_dict[item_name][0]
max_growth = self.growth_values_dict[item_name][1]
if growthLevel > 4:
growthLevel = 4
current_growth_level = self.kh2_read_short(self.Save + slot)
ability = current_growth_level & 0x0FFF
# if the player should be getting a growth ability
if ability | 0x8000 != 0x8000 + min_growth - 1 + growthLevel:
# if it should be level one of that growth
if 0x8000 + min_growth - 1 + growthLevel <= 0x8000 + min_growth or ability < min_growth:
self.kh2_write_short(self.Save + slot, min_growth)
# if it is already in the inventory
elif ability | 0x8000 < (0x8000 + max_growth):
self.kh2_write_short(self.Save + slot, current_growth_level + 1)
for item_name in master_bitmask:
item_data = self.item_name_to_data[item_name]
itemMemory = self.kh2_read_byte(self.Save + item_data.memaddr)
if self.kh2_read_byte(self.Save + item_data.memaddr) & 0x1 << item_data.bitmask == 0:
# when getting a form anti points should be reset to 0 but bit-shift doesn't trigger the game.
if item_name in {"Valor Form", "Wisdom Form", "Limit Form", "Master Form", "Final Form"}:
self.kh2_write_byte(self.Save + 0x3410, 0)
self.kh2_write_byte(self.Save + item_data.memaddr, itemMemory | 0x01 << item_data.bitmask)
for item_name in master_equipment:
item_data = self.item_name_to_data[item_name]
is_there = False
if item_name in self.accessories_set:
Equipment_Anchor_List = self.Equipment_Anchor_Dict["Accessories"]
else:
Equipment_Anchor_List = self.Equipment_Anchor_Dict["Armor"]
# Checking form anchors for the equipment
for slot in Equipment_Anchor_List:
if self.kh2_read_short(self.Save + slot) == item_data.kh2id:
is_there = True
if self.kh2_read_byte(self.Save + item_data.memaddr) != 0:
self.kh2_write_byte(self.Save + item_data.memaddr, 0)
break
if not is_there and item_name not in self.kh2_seed_save["SoldEquipment"]:
if self.kh2_read_byte(self.Save + item_data.memaddr) != 1:
self.kh2_write_byte(self.Save + item_data.memaddr, 1)
for item_name in master_magic:
item_data = self.item_name_to_data[item_name]
amount_of_items = 0
amount_of_items += self.kh2_seed_save_cache["AmountInvo"]["Magic"][item_name]
if self.kh2_read_byte(self.Save + item_data.memaddr) != amount_of_items and self.kh2_read_byte(0x741320) in {10, 8}:
self.kh2_write_byte(self.Save + item_data.memaddr, amount_of_items)
for item_name in master_stat:
amount_of_items = 0
amount_of_items += self.kh2_seed_save_cache["AmountInvo"]["StatIncrease"][item_name]
if self.kh2_read_byte(self.Slot1 + 0x1B2) >= 5:
if item_name == ItemName.MaxHPUp:
if self.kh2_read_byte(self.Save + 0x2498) < 3: # Non-Critical
Bonus = 5
else: # Critical
Bonus = 2
if self.kh2_read_int(self.Slot1 + 0x004) != self.base_hp + (Bonus * amount_of_items):
self.kh2_write_int(self.Slot1 + 0x004, self.base_hp + (Bonus * amount_of_items))
elif item_name == ItemName.MaxMPUp:
if self.kh2_read_byte(self.Save + 0x2498) < 3: # Non-Critical
Bonus = 10
else: # Critical
Bonus = 5
if self.kh2_read_int(self.Slot1 + 0x184) != self.base_mp + (Bonus * amount_of_items):
self.kh2_write_int(self.Slot1 + 0x184, self.base_mp + (Bonus * amount_of_items))
elif item_name == ItemName.DriveGaugeUp:
current_max_drive = self.kh2_read_byte(self.Slot1 + 0x1B2)
# change when max drive is changed from 6 to 4
if current_max_drive < 9 and current_max_drive != self.base_drive + amount_of_items:
self.kh2_write_byte(self.Slot1 + 0x1B2, self.base_drive + amount_of_items)
elif item_name == ItemName.AccessorySlotUp:
current_accessory = self.kh2_read_byte(self.Save + 0x2501)
if current_accessory != self.base_accessory_slots + amount_of_items:
if 4 > current_accessory < self.base_accessory_slots + amount_of_items:
self.kh2_write_byte(self.Save + 0x2501, current_accessory + 1)
elif self.base_accessory_slots + amount_of_items < 4:
self.kh2_write_byte(self.Save + 0x2501, self.base_accessory_slots + amount_of_items)
elif item_name == ItemName.ArmorSlotUp:
current_armor_slots = self.kh2_read_byte(self.Save + 0x2500)
if current_armor_slots != self.base_armor_slots + amount_of_items:
if 4 > current_armor_slots < self.base_armor_slots + amount_of_items:
self.kh2_write_byte(self.Save + 0x2500, current_armor_slots + 1)
elif self.base_armor_slots + amount_of_items < 4:
self.kh2_write_byte(self.Save + 0x2500, self.base_armor_slots + amount_of_items)
elif item_name == ItemName.ItemSlotUp:
current_item_slots = self.kh2_read_byte(self.Save + 0x2502)
if current_item_slots != self.base_item_slots + amount_of_items:
if 8 > current_item_slots < self.base_item_slots + amount_of_items:
self.kh2_write_byte(self.Save + 0x2502, current_item_slots + 1)
elif self.base_item_slots + amount_of_items < 8:
self.kh2_write_byte(self.Save + 0x2502, self.base_item_slots + amount_of_items)
# if self.kh2_read_byte(self.Save + item_data.memaddr) != amount_of_items \
# and self.kh2_read_byte(self.Slot1 + 0x1B2) >= 5 and \
# self.kh2_read_byte(self.Save + 0x23DF) & 0x1 << 3 > 0 and self.kh2_read_byte(0x741320) in {10, 8}:
# self.kh2_write_byte(self.Save + item_data.memaddr, amount_of_items)
if "PoptrackerVersionCheck" in self.kh2slotdata:
if self.kh2slotdata["PoptrackerVersionCheck"] > 4.2 and self.kh2_read_byte(self.Save + 0x3607) != 1: # telling the goa they are on version 4.3
self.kh2_write_byte(self.Save + 0x3607, 1)
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info(e)
logger.info("line 840")
def finishedGame(ctx: KH2Context, message):
if ctx.kh2slotdata['FinalXemnas'] == 1:
if not ctx.final_xemnas and ctx.kh2_read_byte(ctx.Save + all_world_locations[LocationName.FinalXemnas].addrObtained) \
& 0x1 << all_world_locations[LocationName.FinalXemnas].bitIndex > 0:
ctx.final_xemnas = True
# three proofs
if ctx.kh2slotdata['Goal'] == 0:
if ctx.kh2_read_byte(ctx.Save + 0x36B2) > 0 \
and ctx.kh2_read_byte(ctx.Save + 0x36B3) > 0 \
and ctx.kh2_read_byte(ctx.Save + 0x36B4) > 0:
if ctx.kh2slotdata['FinalXemnas'] == 1:
if ctx.final_xemnas:
return True
return False
return True
return False
elif ctx.kh2slotdata['Goal'] == 1:
if ctx.kh2_read_byte(ctx.Save + 0x3641) >= ctx.kh2slotdata['LuckyEmblemsRequired']:
if ctx.kh2_read_byte(ctx.Save + 0x36B3) < 1:
ctx.kh2_write_byte(ctx.Save + 0x36B2, 1)
ctx.kh2_write_byte(ctx.Save + 0x36B3, 1)
ctx.kh2_write_byte(ctx.Save + 0x36B4, 1)
logger.info("The Final Door is now Open")
if ctx.kh2slotdata['FinalXemnas'] == 1:
if ctx.final_xemnas:
return True
return False
return True
return False
elif ctx.kh2slotdata['Goal'] == 2:
# for backwards compat
if "hitlist" in ctx.kh2slotdata:
for boss in ctx.kh2slotdata["hitlist"]:
if boss in message[0]["locations"]:
ctx.hitlist_bounties += 1
if ctx.hitlist_bounties >= ctx.kh2slotdata["BountyRequired"] or ctx.kh2_seed_save_cache["AmountInvo"]["Amount"]["Bounty"] >= ctx.kh2slotdata["BountyRequired"]:
if ctx.kh2_read_byte(ctx.Save + 0x36B3) < 1:
ctx.kh2_write_byte(ctx.Save + 0x36B2, 1)
ctx.kh2_write_byte(ctx.Save + 0x36B3, 1)
ctx.kh2_write_byte(ctx.Save + 0x36B4, 1)
logger.info("The Final Door is now Open")
if ctx.kh2slotdata['FinalXemnas'] == 1:
if ctx.final_xemnas:
return True
return False
return True
return False
elif ctx.kh2slotdata["Goal"] == 3:
if ctx.kh2_seed_save_cache["AmountInvo"]["Amount"]["Bounty"] >= ctx.kh2slotdata["BountyRequired"] and \
ctx.kh2_read_byte(ctx.Save + 0x3641) >= ctx.kh2slotdata['LuckyEmblemsRequired']:
if ctx.kh2_read_byte(ctx.Save + 0x36B3) < 1:
ctx.kh2_write_byte(ctx.Save + 0x36B2, 1)
ctx.kh2_write_byte(ctx.Save + 0x36B3, 1)
ctx.kh2_write_byte(ctx.Save + 0x36B4, 1)
logger.info("The Final Door is now Open")
if ctx.kh2slotdata['FinalXemnas'] == 1:
if ctx.final_xemnas:
return True
return False
return True
return False
async def kh2_watcher(ctx: KH2Context):
while not ctx.exit_event.is_set():
try:
if ctx.kh2connected and ctx.serverconneced:
ctx.sending = []
await asyncio.create_task(ctx.checkWorldLocations())
await asyncio.create_task(ctx.checkLevels())
await asyncio.create_task(ctx.checkSlots())
await asyncio.create_task(ctx.verifyChests())
await asyncio.create_task(ctx.verifyItems())
await asyncio.create_task(ctx.verifyLevel())
message = [{"cmd": 'LocationChecks', "locations": ctx.sending}]
if finishedGame(ctx, message) and not ctx.kh2_finished_game:
await ctx.send_msgs([{"cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL}])
ctx.kh2_finished_game = True
await ctx.send_msgs(message)
elif not ctx.kh2connected and ctx.serverconneced:
logger.info("Game Connection lost. waiting 15 seconds until trying to reconnect.")
ctx.kh2 = None
while not ctx.kh2connected and ctx.serverconneced:
await asyncio.sleep(15)
ctx.kh2 = pymem.Pymem(process_name="KINGDOM HEARTS II FINAL MIX")
if ctx.kh2 is not None:
logger.info("You are now auto-tracking")
ctx.kh2connected = True
except Exception as e:
if ctx.kh2connected:
ctx.kh2connected = False
logger.info(e)
logger.info("line 940")
await asyncio.sleep(0.5)
def launch():
async def main(args):
ctx = KH2Context(args.connect, args.password)
ctx.server_task = asyncio.create_task(server_loop(ctx), name="server loop")
if gui_enabled:
ctx.run_gui()
ctx.run_cli()
progression_watcher = asyncio.create_task(
kh2_watcher(ctx), name="KH2ProgressionWatcher")
await ctx.exit_event.wait()
ctx.server_address = None
await progression_watcher
await ctx.shutdown()
import colorama
parser = get_base_parser(description="KH2 Client, for text interfacing.")
args, rest = parser.parse_known_args()
colorama.init()
asyncio.run(main(args))
colorama.deinit()