From 23f0b720de2337af5114446ee3bc93b03bd9c649 Mon Sep 17 00:00:00 2001 From: qwint Date: Fri, 25 Jul 2025 21:18:36 -0500 Subject: [PATCH] CommonClient: update commands to function without local apworld (#3045) --- CommonClient.py | 164 ++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 130 insertions(+), 34 deletions(-) diff --git a/CommonClient.py b/CommonClient.py index 454150ac..bd7113cb 100644 --- a/CommonClient.py +++ b/CommonClient.py @@ -21,7 +21,7 @@ import Utils if __name__ == "__main__": Utils.init_logging("TextClient", exception_logger="Client") -from MultiServer import CommandProcessor +from MultiServer import CommandProcessor, mark_raw from NetUtils import (Endpoint, decode, NetworkItem, encode, JSONtoTextParser, ClientStatus, Permission, NetworkSlot, RawJSONtoTextParser, add_json_text, add_json_location, add_json_item, JSONTypes, HintStatus, SlotType) from Utils import Version, stream_input, async_start @@ -99,6 +99,17 @@ class ClientCommandProcessor(CommandProcessor): self.ctx.on_print_json({"data": parts, "cmd": "PrintJSON"}) return True + def get_current_datapackage(self) -> dict[str, typing.Any]: + """ + Return datapackage for current game if known. + + :return: The datapackage for the currently registered game. If not found, an empty dictionary will be returned. + """ + if not self.ctx.game: + return {} + checksum = self.ctx.checksums[self.ctx.game] + return Utils.load_data_package_for_checksum(self.ctx.game, checksum) + def _cmd_missing(self, filter_text = "") -> bool: """List all missing location checks, from your local game state. Can be given text, which will be used as filter.""" @@ -107,7 +118,9 @@ class ClientCommandProcessor(CommandProcessor): return False count = 0 checked_count = 0 - for location, location_id in AutoWorldRegister.world_types[self.ctx.game].location_name_to_id.items(): + + lookup = self.get_current_datapackage().get("location_name_to_id", {}) + for location, location_id in lookup.items(): if filter_text and filter_text not in location: continue if location_id < 0: @@ -128,43 +141,91 @@ class ClientCommandProcessor(CommandProcessor): self.output("No missing location checks found.") return True - def _cmd_items(self): + def output_datapackage_part(self, key: str, name: str) -> bool: + """ + Helper to digest a specific section of this game's datapackage. + + :param key: The dictionary key in the datapackage. + :param name: Printed to the user as context for the part. + + :return: Whether the process was successful. + """ + if not self.ctx.game: + self.output(f"No game set, cannot determine {name}.") + return False + + lookup = self.get_current_datapackage().get(key) + if lookup is None: + self.output("datapackage not yet loaded, try again") + return False + + self.output(f"{name} for {self.ctx.game}") + for key in lookup: + self.output(key) + return True + + def _cmd_items(self) -> bool: """List all item names for the currently running game.""" - if not self.ctx.game: - self.output("No game set, cannot determine existing items.") - return False - self.output(f"Item Names for {self.ctx.game}") - for item_name in AutoWorldRegister.world_types[self.ctx.game].item_name_to_id: - self.output(item_name) + return self.output_datapackage_part("item_name_to_id", "Item Names") - def _cmd_item_groups(self): - """List all item group names for the currently running game.""" - if not self.ctx.game: - self.output("No game set, cannot determine existing item groups.") - return False - self.output(f"Item Group Names for {self.ctx.game}") - for group_name in AutoWorldRegister.world_types[self.ctx.game].item_name_groups: - self.output(group_name) - - def _cmd_locations(self): + def _cmd_locations(self) -> bool: """List all location names for the currently running game.""" - if not self.ctx.game: - self.output("No game set, cannot determine existing locations.") - return False - self.output(f"Location Names for {self.ctx.game}") - for location_name in AutoWorldRegister.world_types[self.ctx.game].location_name_to_id: - self.output(location_name) + return self.output_datapackage_part("location_name_to_id", "Location Names") - def _cmd_location_groups(self): - """List all location group names for the currently running game.""" - if not self.ctx.game: - self.output("No game set, cannot determine existing location groups.") - return False - self.output(f"Location Group Names for {self.ctx.game}") - for group_name in AutoWorldRegister.world_types[self.ctx.game].location_name_groups: - self.output(group_name) + def output_group_part(self, group_key: typing.Literal["item_name_groups", "location_name_groups"], + filter_key: str, + name: str) -> bool: + """ + Logs an item or location group from the player's game's datapackage. - def _cmd_ready(self): + :param group_key: Either Item or Location group to be processed. + :param filter_key: Which group key to filter to. If an empty string is passed will log all item/location groups. + :param name: Printed to the user as context for the part. + + :return: Whether the process was successful. + """ + if not self.ctx.game: + self.output(f"No game set, cannot determine existing {name} Groups.") + return False + lookup = Utils.persistent_load().get("groups_by_checksum", {}).get(self.ctx.checksums[self.ctx.game], {})\ + .get(self.ctx.game, {}).get(group_key, {}) + if lookup is None: + self.output("datapackage not yet loaded, try again") + return False + + if filter_key: + if filter_key not in lookup: + self.output(f"Unknown {name} Group {filter_key}") + return False + + self.output(f"{name}s for {name} Group \"{filter_key}\"") + for entry in lookup[filter_key]: + self.output(entry) + else: + self.output(f"{name} Groups for {self.ctx.game}") + for group in lookup: + self.output(group) + return True + + @mark_raw + def _cmd_item_groups(self, key: str = "") -> bool: + """ + List all item group names for the currently running game. + + :param key: Which item group to filter to. Will log all groups if empty. + """ + return self.output_group_part("item_name_groups", key, "Item") + + @mark_raw + def _cmd_location_groups(self, key: str = "") -> bool: + """ + List all location group names for the currently running game. + + :param key: Which item group to filter to. Will log all groups if empty. + """ + return self.output_group_part("location_name_groups", key, "Location") + + def _cmd_ready(self) -> bool: """Send ready status to server.""" self.ctx.ready = not self.ctx.ready if self.ctx.ready: @@ -174,6 +235,7 @@ class ClientCommandProcessor(CommandProcessor): state = ClientStatus.CLIENT_CONNECTED self.output("Unreadied.") async_start(self.ctx.send_msgs([{"cmd": "StatusUpdate", "status": state}]), name="send StatusUpdate") + return True def default(self, raw: str): """The default message parser to be used when parsing any messages that do not match a command""" @@ -379,6 +441,8 @@ class CommonContext: self.jsontotextparser = JSONtoTextParser(self) self.rawjsontotextparser = RawJSONtoTextParser(self) + if self.game: + self.checksums[self.game] = network_data_package["games"][self.game]["checksum"] self.update_data_package(network_data_package) # execution @@ -638,6 +702,24 @@ class CommonContext: for game, game_data in data_package["games"].items(): Utils.store_data_package_for_checksum(game, game_data) + def consume_network_item_groups(self): + data = {"item_name_groups": self.stored_data[f"_read_item_name_groups_{self.game}"]} + current_cache = Utils.persistent_load().get("groups_by_checksum", {}).get(self.checksums[self.game], {}) + if self.game in current_cache: + current_cache[self.game].update(data) + else: + current_cache[self.game] = data + Utils.persistent_store("groups_by_checksum", self.checksums[self.game], current_cache) + + def consume_network_location_groups(self): + data = {"location_name_groups": self.stored_data[f"_read_location_name_groups_{self.game}"]} + current_cache = Utils.persistent_load().get("groups_by_checksum", {}).get(self.checksums[self.game], {}) + if self.game in current_cache: + current_cache[self.game].update(data) + else: + current_cache[self.game] = data + Utils.persistent_store("groups_by_checksum", self.checksums[self.game], current_cache) + # data storage def set_notify(self, *keys: str) -> None: @@ -938,6 +1020,12 @@ async def process_server_cmd(ctx: CommonContext, args: dict): ctx.hint_points = args.get("hint_points", 0) ctx.consume_players_package(args["players"]) ctx.stored_data_notification_keys.add(f"_read_hints_{ctx.team}_{ctx.slot}") + if ctx.game: + game = ctx.game + else: + game = ctx.slot_info[ctx.slot][1] + ctx.stored_data_notification_keys.add(f"_read_item_name_groups_{game}") + ctx.stored_data_notification_keys.add(f"_read_location_name_groups_{game}") msgs = [] if ctx.locations_checked: msgs.append({"cmd": "LocationChecks", @@ -1018,11 +1106,19 @@ async def process_server_cmd(ctx: CommonContext, args: dict): ctx.stored_data.update(args["keys"]) if ctx.ui and f"_read_hints_{ctx.team}_{ctx.slot}" in args["keys"]: ctx.ui.update_hints() + if f"_read_item_name_groups_{ctx.game}" in args["keys"]: + ctx.consume_network_item_groups() + if f"_read_location_name_groups_{ctx.game}" in args["keys"]: + ctx.consume_network_location_groups() elif cmd == "SetReply": ctx.stored_data[args["key"]] = args["value"] if ctx.ui and f"_read_hints_{ctx.team}_{ctx.slot}" == args["key"]: ctx.ui.update_hints() + elif f"_read_item_name_groups_{ctx.game}" == args["key"]: + ctx.consume_network_item_groups() + elif f"_read_location_name_groups_{ctx.game}" == args["key"]: + ctx.consume_network_location_groups() elif args["key"].startswith("EnergyLink"): ctx.current_energy_link_value = args["value"] if ctx.ui: