KH2: Client Optimizations and some QoL (#4547)

* adding qwints suggestions

* add stat increase protection and ingame yml stuff

* idk how I forgot these

* reword things

* Update worlds/kh2/Client.py

Co-authored-by: qwint <qwint.42@gmail.com>

* 3.12 compat

* too long of a line

* why didnt I do this before lol

* reading is hard

* missed one

* forgot the self

* fix crash if you get datapackage that isnt kh2

* update to main?

* update to use 0.10 as base and fix violet's base 0 on hex values

* reverting this because I'm bad at my job

---------

Co-authored-by: qwint <qwint.42@gmail.com>
This commit is contained in:
JaredWeakStrike
2025-03-08 08:58:59 -05:00
committed by GitHub
parent 0f738935ee
commit 3f8e3082c0
2 changed files with 136 additions and 69 deletions

View File

@@ -1,4 +1,5 @@
import ModuleUpdate
import Utils
ModuleUpdate.update()
@@ -23,6 +24,7 @@ class KH2Context(CommonContext):
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
@@ -35,6 +37,7 @@ class KH2Context(CommonContext):
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_data_package = {}
self.kh2_loc_name_to_id = None
self.kh2_item_name_to_id = None
self.lookup_id_to_item = None
@@ -83,6 +86,8 @@ class KH2Context(CommonContext):
},
}
self.kh2seedname = None
self.kh2_seed_save_path_join = None
self.kh2slotdata = None
self.mem_json = None
self.itemamount = {}
@@ -114,26 +119,18 @@ class KH2Context(CommonContext):
# 255: {}, # starting screen
}
self.last_world_int = -1
# 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 old address
# epic addresses
# epic .10 addresses
self.Now = 0x0716DF8
self.Save = 0x09A92F0
self.Save = 0x9A9330
self.Journal = 0x743260
self.Shop = 0x743350
self.Slot1 = 0x2A22FD8
# self.Sys3 = 0x2A59DF0
# self.Bt10 = 0x2A74880
# self.BtlEnd = 0x2A0D3E0
# self.Slot1 = 0x2A20C98 old address
self.Slot1 = 0x2A23018
self.kh2_game_version = None # can be egs or steam
self.kh2_seed_save_path = None
self.chest_set = set(exclusion_table["Chests"])
self.keyblade_set = set(CheckDupingItems["Weapons"]["Keyblades"])
self.staff_set = set(CheckDupingItems["Weapons"]["Staffs"])
@@ -194,8 +191,7 @@ class KH2Context(CommonContext):
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:
with open(self.kh2_seed_save_path_join, 'w') as f:
f.write(json.dumps(self.kh2_seed_save, indent=4))
await super(KH2Context, self).connection_closed()
@@ -203,8 +199,7 @@ class KH2Context(CommonContext):
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:
with open(self.kh2_seed_save_path_join, 'w') as f:
f.write(json.dumps(self.kh2_seed_save, indent=4))
await super(KH2Context, self).disconnect()
@@ -217,8 +212,7 @@ class KH2Context(CommonContext):
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:
with open(self.kh2_seed_save_path_join, 'w') as f:
f.write(json.dumps(self.kh2_seed_save, indent=4))
await super(KH2Context, self).shutdown()
@@ -232,7 +226,7 @@ class KH2Context(CommonContext):
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")
return int.from_bytes(self.kh2.read_bytes(self.kh2.base_address + address, 1))
def kh2_read_int(self, address):
return self.kh2.read_int(self.kh2.base_address + address)
@@ -244,11 +238,14 @@ class KH2Context(CommonContext):
return self.kh2.read_string(self.kh2.base_address + address, length)
def on_package(self, cmd: str, args: dict):
if cmd in {"RoomInfo"}:
if cmd == "RoomInfo":
self.kh2seedname = args['seed_name']
self.kh2_seed_save_path = f"kh2save2{self.kh2seedname}{self.auth}.json"
self.kh2_seed_save_path_join = os.path.join(self.game_communication_path, self.kh2_seed_save_path)
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"):
if not os.path.exists(self.kh2_seed_save_path_join):
self.kh2_seed_save = {
"Levels": {
"SoraLevel": 0,
@@ -261,12 +258,11 @@ class KH2Context(CommonContext):
},
"SoldEquipment": [],
}
with open(os.path.join(self.game_communication_path, f"kh2save2{self.kh2seedname}{self.auth}.json"),
'wt') as f:
with open(self.kh2_seed_save_path_join, '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:
elif os.path.exists(self.kh2_seed_save_path_join):
with open(self.kh2_seed_save_path_join) as f:
self.kh2_seed_save = json.load(f)
if self.kh2_seed_save is None:
self.kh2_seed_save = {
@@ -284,13 +280,22 @@ class KH2Context(CommonContext):
# 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"]}]))
if cmd == "Connected":
self.kh2slotdata = args['slot_data']
# self.kh2_local_items = {int(location): item for location, item in self.kh2slotdata["LocalItems"].items()}
self.kh2_data_package = Utils.load_data_package_for_checksum(
"Kingdom Hearts 2", self.checksums["Kingdom Hearts 2"])
if "location_name_to_id" in self.kh2_data_package:
self.data_package_kh2_cache(
self.kh2_data_package["location_name_to_id"], self.kh2_data_package["item_name_to_id"])
self.connect_to_game()
else:
asyncio.create_task(self.send_msgs([{"cmd": "GetDataPackage", "games": ["Kingdom Hearts 2"]}]))
self.locations_checked = set(args["checked_locations"])
if cmd in {"ReceivedItems"}:
if cmd == "ReceivedItems":
# 0x2546
# 0x2658
# 0x276A
@@ -338,42 +343,44 @@ class KH2Context(CommonContext):
for item in args['items']:
asyncio.create_task(self.give_item(item.item, item.location))
if cmd in {"RoomUpdate"}:
if cmd == "RoomUpdate":
if "checked_locations" in args:
new_locations = set(args["checked_locations"])
self.locations_checked |= new_locations
if cmd in {"DataPackage"}:
if cmd == "DataPackage":
if "Kingdom Hearts 2" in args["data"]["games"]:
self.data_package_kh2_cache(args)
if "KeybladeAbilities" in self.kh2slotdata.keys():
# sora ability to slot
self.AbilityQuantityDict.update(self.kh2slotdata["KeybladeAbilities"])
# itemid:[slots that are available for that item]
self.AbilityQuantityDict.update(self.kh2slotdata["StaffAbilities"])
self.AbilityQuantityDict.update(self.kh2slotdata["ShieldAbilities"])
self.data_package_kh2_cache(
args["data"]["games"]["Kingdom Hearts 2"]["location_name_to_id"],
args["data"]["games"]["Kingdom Hearts 2"]["item_name_to_id"])
self.connect_to_game()
asyncio.create_task(self.send_msgs([{'cmd': 'Sync'}]))
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)
def connect_to_game(self):
if "KeybladeAbilities" in self.kh2slotdata.keys():
# sora ability to slot
self.AbilityQuantityDict.update(self.kh2slotdata["KeybladeAbilities"])
# itemid:[slots that are available for that item]
self.AbilityQuantityDict.update(self.kh2slotdata["StaffAbilities"])
self.AbilityQuantityDict.update(self.kh2slotdata["ShieldAbilities"])
try:
if not self.kh2:
self.kh2 = pymem.Pymem(process_name="KINGDOM HEARTS II FINAL MIX")
self.get_addresses()
self.all_weapon_location_id = {self.kh2_loc_name_to_id[loc] for loc in all_weapon_slot}
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'}]))
try:
if not self.kh2:
self.kh2 = pymem.Pymem(process_name="KINGDOM HEARTS II FINAL MIX")
self.get_addresses()
def data_package_kh2_cache(self, args):
self.kh2_loc_name_to_id = args["data"]["games"]["Kingdom Hearts 2"]["location_name_to_id"]
except Exception as e:
if self.kh2connected:
self.kh2connected = False
logger.info("Game is not open.")
self.serverconneced = True
def data_package_kh2_cache(self, loc_to_id, item_to_id):
self.kh2_loc_name_to_id = loc_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.kh2_item_name_to_id = item_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"]]
@@ -742,7 +749,8 @@ class KH2Context(CommonContext):
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:
# checking if they talked to the computer to give them these
if self.kh2_read_byte(self.Slot1 + 0x1B2) >= 5 and (self.kh2_read_byte(self.Save + 0x1D27) & 0x1 << 3) > 0:
if item_name == ItemName.MaxHPUp:
if self.kh2_read_byte(self.Save + 0x2498) < 3: # Non-Critical
Bonus = 5
@@ -808,34 +816,33 @@ class KH2Context(CommonContext):
def get_addresses(self):
if not self.kh2connected and self.kh2 is not None:
if self.kh2_game_version is None:
if self.kh2_read_string(0x09A9830, 4) == "KH2J":
# current verions is .10 then runs the get from github stuff
if self.kh2_read_string(0x9A98B0, 4) == "KH2J":
self.kh2_game_version = "STEAM"
self.Now = 0x0717008
self.Save = 0x09A9830
self.Slot1 = 0x2A23518
self.Save = 0x09A98B0
self.Slot1 = 0x2A23598
self.Journal = 0x7434E0
self.Shop = 0x7435D0
elif self.kh2_read_string(0x09A92F0, 4) == "KH2J":
elif self.kh2_read_string(0x9A9330, 4) == "KH2J":
self.kh2_game_version = "EGS"
else:
if self.game_communication_path:
logger.info("Checking with most up to date addresses of github. If file is not found will be downloading datafiles. This might take a moment")
logger.info("Checking with most up to date addresses from the addresses json.")
#if mem addresses file is found then check version and if old get new one
kh2memaddresses_path = os.path.join(self.game_communication_path, f"kh2memaddresses.json")
kh2memaddresses_path = os.path.join(self.game_communication_path, "kh2memaddresses.json")
if not os.path.exists(kh2memaddresses_path):
logger.info("File is not found. Downloading json with memory addresses. This might take a moment")
mem_resp = requests.get("https://raw.githubusercontent.com/JaredWeakStrike/KH2APMemoryValues/master/kh2memaddresses.json")
if mem_resp.status_code == 200:
self.mem_json = json.loads(mem_resp.content)
with open(kh2memaddresses_path,
'w') as f:
with open(kh2memaddresses_path, 'w') as f:
f.write(json.dumps(self.mem_json, indent=4))
else:
with open(kh2memaddresses_path, 'r') as f:
with open(kh2memaddresses_path) as f:
self.mem_json = json.load(f)
if self.mem_json:
for key in self.mem_json.keys():
if self.kh2_read_string(int(self.mem_json[key]["GameVersionCheck"], 0), 4) == "KH2J":
self.Now = int(self.mem_json[key]["Now"], 0)
self.Save = int(self.mem_json[key]["Save"], 0)

View File

@@ -368,6 +368,37 @@ def patch_kh2(self, output_directory):
}
]
},
{
'name': 'msg/us/he.bar',
'multi': [
{
'name': 'msg/fr/he.bar'
},
{
'name': 'msg/gr/he.bar'
},
{
'name': 'msg/it/he.bar'
},
{
'name': 'msg/sp/he.bar'
}
],
'method': 'binarc',
'source': [
{
'name': 'he',
'type': 'list',
'method': 'kh2msg',
'source': [
{
'name': 'he.yml',
'language': 'en'
}
]
}
]
},
],
'title': 'Randomizer Seed'
}
@@ -411,6 +442,34 @@ def patch_kh2(self, output_directory):
'en': f"Your Level Depth is {self.options.LevelDepth.current_option_name}"
}
]
self.fight_and_form_text = [
{
'id': 15121, # poster name
'en': f"Game Options"
},
{
'id': 15122,
'en': f"Fight Logic is {self.options.FightLogic.current_option_name}\n"
f"Auto Form Logic is {self.options.AutoFormLogic.current_option_name}\n"
f"Final Form Logic is {self.options.FinalFormLogic.current_option_name}"
}
]
self.cups_text = [
{
'id': 4043,
'en': f"CupsToggle: {self.options.Cups.current_option_name}"
},
{
'id': 4044,
'en': f"CupsToggle: {self.options.Cups.current_option_name}"
},
{
'id': 4045,
'en': f"CupsToggle: {self.options.Cups.current_option_name}"
},
]
mod_dir = os.path.join(output_directory, mod_name + "_" + Utils.__version__)
self.mod_yml["title"] = f"Randomizer Seed {mod_name}"
@@ -423,7 +482,8 @@ def patch_kh2(self, output_directory):
"FmlvList.yml": yaml.dump(self.formattedFmlv, line_break="\n"),
"mod.yml": yaml.dump(self.mod_yml, line_break="\n"),
"po.yml": yaml.dump(self.pooh_text, line_break="\n"),
"sys.yml": yaml.dump(self.level_depth_text, line_break="\n"),
"sys.yml": yaml.dump(self.level_depth_text + self.fight_and_form_text, line_break="\n"),
"he.yml": yaml.dump(self.cups_text, line_break="\n")
}
mod = KH2Container(openkhmod, mod_dir, output_directory, self.player,