Kingdom Hearts RE Chain of Memories: Implement New Game

KHRECOMClient.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+import ModuleUpdate
+ModuleUpdate.update()
+
+from worlds.khrecom.Client import check_stdin, launch
+import Utils
+
+if __name__ == "__main__":
+    Utils.init_logging("KHRECOMClient", exception_logger="Client")
+    check_stdin()
+    launch()

README.md

@@ -75,6 +75,7 @@ Currently, the following games are supported:
 * Old School Runescape
 * Kingdom Hearts 1
 * Mega Man 2
+* Kingdom Hearts RE: Chain of Memories
 
 For setup and instructions check out our [tutorials page](https://archipelago.gg/tutorial/).
 Downloads can be found at [Releases](https://github.com/ArchipelagoMW/Archipelago/releases), including compiled

docs/CODEOWNERS

@@ -81,6 +81,9 @@
 # Kingdom Hearts
 /worlds/kh1/ @gaithern
 
+# Kingdom Hearts RE: Chain of Memories
+/worlds/khrecom/ @gaithern
+
 # Kingdom Hearts 2
 /worlds/kh2/ @JaredWeakStrike
 

worlds/khrecom/Client.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+import os
+import sys
+import asyncio
+import shutil
+import logging
+
+import ModuleUpdate
+ModuleUpdate.update()
+
+import Utils
+
+item_num = 1
+
+logger = logging.getLogger("Client")
+
+if __name__ == "__main__":
+    Utils.init_logging("KHRECOMClient", exception_logger="Client")
+
+from NetUtils import NetworkItem, ClientStatus
+from CommonClient import gui_enabled, logger, get_base_parser, ClientCommandProcessor, \
+    CommonContext, server_loop
+
+
+def check_stdin() -> None:
+    if Utils.is_windows and sys.stdin:
+        print("WARNING: Console input is not routed reliably on Windows, use the GUI instead.")
+
+class KHRECOMClientCommandProcessor(ClientCommandProcessor):
+    pass
+
+
+class KHRECOMContext(CommonContext):
+    command_processor: int = KHRECOMClientCommandProcessor
+    game = "Kingdom Hearts RE Chain of Memories"
+    items_handling = 0b111  # full remote
+
+    def __init__(self, server_address, password):
+        super(KHRECOMContext, self).__init__(server_address, password)
+        self.send_index: int = 0
+        self.syncing = False
+        self.awaiting_bridge = False
+        # self.game_communication_path: files go in this path to pass data between us and the actual game
+        if "localappdata" in os.environ:
+            self.game_communication_path = os.path.expandvars(r"%localappdata%/KHRECOM")
+        else:
+            self.game_communication_path = os.path.expandvars(r"$HOME/KHRECOM")
+        if not os.path.exists(self.game_communication_path):
+            os.makedirs(self.game_communication_path)
+        for root, dirs, files in os.walk(self.game_communication_path):
+            for file in files:
+                if file.find("obtain") <= -1:
+                    os.remove(root+"/"+file)
+
+    async def server_auth(self, password_requested: bool = False):
+        if password_requested and not self.password:
+            await super(KHRECOMContext, self).server_auth(password_requested)
+        await self.get_username()
+        await self.send_connect()
+
+    async def connection_closed(self):
+        await super(KHRECOMContext, self).connection_closed()
+        for root, dirs, files in os.walk(self.game_communication_path):
+            for file in files:
+                if file.find("obtain") <= -1:
+                    os.remove(root + "/" + file)
+        global item_num
+        item_num = 1
+
+    @property
+    def endpoints(self):
+        if self.server:
+            return [self.server]
+        else:
+            return []
+
+    async def shutdown(self):
+        await super(KHRECOMContext, self).shutdown()
+        for root, dirs, files in os.walk(self.game_communication_path):
+            for file in files:
+                if file.find("obtain") <= -1:
+                    os.remove(root+"/"+file)
+        global item_num
+        item_num = 1
+
+    def on_package(self, cmd: str, args: dict):
+        if cmd in {"Connected"}:
+            if not os.path.exists(self.game_communication_path):
+                os.makedirs(self.game_communication_path)
+            for ss in self.checked_locations:
+                filename = f"send{ss}"
+                with open(os.path.join(self.game_communication_path, filename), 'w') as f:
+                    f.close()
+            for key in list(args['slot_data'].keys()):
+                with open(os.path.join(self.game_communication_path, key + ".cfg"), 'w') as f:
+                    f.write(str(args['slot_data'][key]))
+                    f.close()
+        if cmd in {"ReceivedItems"}:
+            start_index = args["index"]
+            if start_index != len(self.items_received):
+                global item_num
+                for item in args['items']:
+                    found = False
+                    item_filename = f"AP_{str(item_num)}.item"
+                    for filename in os.listdir(self.game_communication_path):
+                        if filename == item_filename:
+                            found = True
+                    if not found:
+                        with open(os.path.join(self.game_communication_path, item_filename), 'w') as f:
+                            f.write(str(NetworkItem(*item).item) + "\n" + str(NetworkItem(*item).location) + "\n" + str(NetworkItem(*item).player))
+                            f.close()
+                            item_num = item_num + 1
+
+        if cmd in {"RoomUpdate"}:
+            if "checked_locations" in args:
+                for ss in self.checked_locations:
+                    filename = f"send{ss}"
+                    with open(os.path.join(self.game_communication_path, filename), 'w') as f:
+                        f.close()
+
+    def run_gui(self):
+        """Import kivy UI system and start running it as self.ui_task."""
+        from kvui import GameManager
+
+        class KHRECOMManager(GameManager):
+            logging_pairs = [
+                ("Client", "Archipelago")
+            ]
+            base_title = "Archipelago KHRECOM Client"
+
+        self.ui = KHRECOMManager(self)
+        self.ui_task = asyncio.create_task(self.ui.async_run(), name="UI")
+
+
+async def game_watcher(ctx: KHRECOMContext):
+    from .Locations import lookup_id_to_name
+    while not ctx.exit_event.is_set():
+        if ctx.syncing == True:
+            sync_msg = [{'cmd': 'Sync'}]
+            if ctx.locations_checked:
+                sync_msg.append({"cmd": "LocationChecks", "locations": list(ctx.locations_checked)})
+            await ctx.send_msgs(sync_msg)
+            ctx.syncing = False
+        sending = []
+        victory = False
+        for root, dirs, files in os.walk(ctx.game_communication_path):
+            for file in files:
+                if file.find("send") > -1:
+                    st = file.split("send", -1)[1]
+                    if st != "nil":
+                        sending = sending+[(int(st))]
+                if file.find("victory") > -1:
+                    victory = True
+        ctx.locations_checked = sending
+        message = [{"cmd": 'LocationChecks', "locations": sending}]
+        await ctx.send_msgs(message)
+        if not ctx.finished_game and victory:
+            await ctx.send_msgs([{"cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL}])
+            ctx.finished_game = True
+        await asyncio.sleep(0.1)
+
+
+def launch():
+    async def main(args):
+        ctx = KHRECOMContext(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(
+            game_watcher(ctx), name="KHRECOMProgressionWatcher")
+
+        await ctx.exit_event.wait()
+        ctx.server_address = None
+
+        await progression_watcher
+
+        await ctx.shutdown()
+
+    import colorama
+
+    parser = get_base_parser(description="KHRECOM Client, for text interfacing.")
+
+    args, rest = parser.parse_known_args()
+    colorama.init()
+    asyncio.run(main(args))
+    colorama.deinit()