KH2: Client Optimizations and some QoL (ArchipelagoMW#4547)

JaredWeakStrike · qwint · Mego · commit ec4bc6932951 · 2025-03-08T17:32:12.000-06:00
* 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 &lt;qwint.42@gmail.com&gt;

* 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 &lt;qwint.42@gmail.com&gt;
diff --git a/worlds/kh2/Client.py b/worlds/kh2/Client.py
@@ -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 @@ def __init__(self, server_address, password):
         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 @@ def __init__(self, server_address, password):
             },
         }
         self.kh2seedname = None
+        self.kh2_seed_save_path_join = None
+
         self.kh2slotdata = None
         self.mem_json = None
         self.itemamount = {}
@@ -114,26 +119,18 @@ def __init__(self, server_address, password):
             #  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,17 +191,15 @@ 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:
+            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()
 
     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:
+            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 @@ def endpoints(self):
 
     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 @@ 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")
+        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 @@ def kh2_read_string(self, address, length):
         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 @@ def on_package(self, cmd: str, args: dict):
                     },
                     "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 @@ def on_package(self, cmd: str, args: dict):
                     # 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 @@ def on_package(self, cmd: str, args: dict):
                 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"])
-
-            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:
-                if not self.kh2:
-                    self.kh2 = pymem.Pymem(process_name="KINGDOM HEARTS II FINAL MIX")
-                    self.get_addresses()
-
-            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'}]))
-
-    def data_package_kh2_cache(self, args):
-        self.kh2_loc_name_to_id = args["data"]["games"]["Kingdom Hearts 2"]["location_name_to_id"]
+                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'}]))
+
+    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"])
+
+        self.all_weapon_location_id = {self.kh2_loc_name_to_id[loc] for loc in all_weapon_slot}
+
+        try:
+            if not self.kh2:
+                self.kh2 = pymem.Pymem(process_name="KINGDOM HEARTS II FINAL MIX")
+                self.get_addresses()
+
+        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 @@ async def verifyItems(self):
             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 @@ async def verifyItems(self):
     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)
diff --git a/worlds/kh2/OpenKH.py b/worlds/kh2/OpenKH.py
@@ -368,6 +368,37 @@ def increaseStat(i):
                     }
                 ]
             },
+            {
+                '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 increaseStat(i):
             '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 increaseStat(i):
         "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,
