Core: Remove Universally Unique ID Requirements (Per-Game Data Packages)

AdventureClient.py

@@ -112,7 +112,7 @@ def on_package(self, cmd: str, args: dict):
             if ': !' not in msg:
                 self._set_message(msg, SYSTEM_MESSAGE_ID)
         elif cmd == "ReceivedItems":
-            msg = f"Received {', '.join([self.item_names[item.item] for item in args['items']])}"
+            msg = f"Received {', '.join([self.item_names[self.game][item.item] for item in args['items']])}"
             self._set_message(msg, SYSTEM_MESSAGE_ID)
         elif cmd == "Retrieved":
             if f"adventure_{self.auth}_freeincarnates_used" in args["keys"]:

CommonClient.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import collections
 import copy
 import logging
 import asyncio
@@ -173,10 +174,82 @@ class CommonContext:
     items_handling: typing.Optional[int] = None
     want_slot_data: bool = True  # should slot_data be retrieved via Connect
 
+    class ImplicitNameLookupDict(collections.abc.MutableMapping):
+        _store: typing.Dict
+
+        def __init__(self, lookup_type: typing.Literal["item", "location"]):
+            self.lookup_type = lookup_type
+            self._archipelago_lookup = {}
+            self._store = collections.defaultdict(lambda: Utils.KeyedDefaultDict(lambda key: f"Unknown {lookup_type} "
+                                                                                             f"(ID {key})"))
+
+        def __getitem__(self, key):
+            if isinstance(key, str):
+                return self._store[key]
+
+            logger.warning(f"Attempting to do an implicit lookup for an {self.lookup_type}. To avoid ambiguous id "
+                           f"names, attempt lookup with the game name. See f{self.lookup_type}_names docstring for "
+                           f"usage examples.")
+            name = self._store.get(key, f"Unknown {self.lookup_type} (ID: {key})")
+            return name if name is not None else f"Ambiguous {self.lookup_type} (ID: {key})"
+
+        def __setitem__(self, key, value):
+            if isinstance(key, int) and key in self._store:
+                # A `None` value signifies a duplicate item/location with the same id already exists. If anyone is
+                # trying to do an implicit name lookup, we can't be sure which game it is, so this will cause
+                # 'Ambiguous item/location' to be returned when `__getitem__` is called.
+                self._store[key] = None
+                return
+
+            self._store[key] = value
+
+        def __delitem__(self, key):
+            del self._store[key]
+
+        def __len__(self):
+            return len(self._store)
+
+        def __iter__(self):
+            return iter(self._store)
+
+        def __repr__(self):
+            return self._store.__repr__()
+
+        def update_game(self, game: str, lookup_table: typing.Dict[str, int]):
+            reverse_lookup_table = Utils.KeyedDefaultDict(lambda key: f"Unknown {self.lookup_type} (ID: {key})")
+            reverse_lookup_table.update({code: name for name, code in lookup_table.items()})
+            self._store[game] = collections.ChainMap(self._archipelago_lookup, reverse_lookup_table)
+            self.update(reverse_lookup_table)
+            if game == "Archipelago":
+                # Keep track of the Archipelago data package separately so if it gets updated in a custom datapackage,
+                # it updates in all chain maps automatically.
+                self._archipelago_lookup.clear()
+                self._archipelago_lookup.update(reverse_lookup_table)
+
     # data package
     # Contents in flux until connection to server is made, to download correct data for this multiworld.
-    item_names: typing.Dict[int, str] = Utils.KeyedDefaultDict(lambda code: f'Unknown item (ID:{code})')
-    location_names: typing.Dict[int, str] = Utils.KeyedDefaultDict(lambda code: f'Unknown location (ID:{code})')
+    item_names: ImplicitNameLookupDict
+    """A dictionary lookup of id -> name for items and updates automatically when a new data package arrives. If game 
+    name is omitted, it will attempt a lookup in all game packages.
+
+    Returns "Unknown item" if an id does not exist in session data package.
+    Returns "Ambiguous item" if a game name is not provided and multiple items share the same id.
+
+    Example usages::
+        name = item_names[game][id]  # Explicit id lookup with game name.
+        name = item_names[id]        # Implicit id lookup. May return "Ambiguous Item" if duplicate ids exist.
+    """
+    location_names: ImplicitNameLookupDict
+    """A dictionary lookup of id -> name for locations and updates automatically when a new data package arrives. If 
+    game name is omitted, it will attempt a lookup in all game packages.
+
+    Returns "Unknown location" if an id does not exist in session data package.
+    Returns "Ambiguous location" if a game name is not provided and multiple locations share the same id.
+
+    Example usages::
+        name = location_names[game][id]  # Explicit id lookup with game name.
+        name = location_names[id]        # Implicit id lookup. May return "Ambiguous Location" if duplicate ids exist.
+    """
 
     # defaults
     starting_reconnect_delay: int = 5
@@ -231,7 +304,7 @@ class CommonContext:
     # message box reporting a loss of connection
     _messagebox_connection_loss: typing.Optional["kvui.MessageBox"] = None
 
-    def __init__(self, server_address: typing.Optional[str], password: typing.Optional[str]) -> None:
+    def __init__(self, server_address: typing.Optional[str] = None, password: typing.Optional[str] = None) -> None:
         # server state
         self.server_address = server_address
         self.username = None
@@ -271,6 +344,9 @@ def __init__(self, server_address: typing.Optional[str], password: typing.Option
         self.exit_event = asyncio.Event()
         self.watcher_event = asyncio.Event()
 
+        self.item_names = self.ImplicitNameLookupDict("item")
+        self.location_names = self.ImplicitNameLookupDict("location")
+
         self.jsontotextparser = JSONtoTextParser(self)
         self.rawjsontotextparser = RawJSONtoTextParser(self)
         self.update_data_package(network_data_package)
@@ -486,19 +562,17 @@ async def prepare_data_package(self, relevant_games: typing.Set[str],
                         or remote_checksum != cache_checksum:
                     needed_updates.add(game)
                 else:
-                    self.update_game(cached_game)
+                    self.update_game(cached_game, game)
         if needed_updates:
             await self.send_msgs([{"cmd": "GetDataPackage", "games": [game_name]} for game_name in needed_updates])
 
-    def update_game(self, game_package: dict):
-        for item_name, item_id in game_package["item_name_to_id"].items():
-            self.item_names[item_id] = item_name
-        for location_name, location_id in game_package["location_name_to_id"].items():
-            self.location_names[location_id] = location_name
+    def update_game(self, game_package: dict, game: str):
+        self.item_names.update_game(game, game_package["item_name_to_id"])
+        self.location_names.update_game(game, game_package["location_name_to_id"])
 
     def update_data_package(self, data_package: dict):
         for game, game_data in data_package["games"].items():
-            self.update_game(game_data)
+            self.update_game(game_data, game)
 
     def consume_network_data_package(self, data_package: dict):
         self.update_data_package(data_package)

MultiServer.py

@@ -168,9 +168,11 @@ class Context:
     slot_info: typing.Dict[int, NetworkSlot]
     generator_version = Version(0, 0, 0)
     checksums: typing.Dict[str, str]
-    item_names: typing.Dict[int, str] = Utils.KeyedDefaultDict(lambda code: f'Unknown item (ID:{code})')
+    item_names: typing.Dict[str, typing.Dict[int, str]] = (
+        collections.defaultdict(lambda: Utils.KeyedDefaultDict(lambda code: f'Unknown item (ID:{code})')))
     item_name_groups: typing.Dict[str, typing.Dict[str, typing.Set[str]]]
-    location_names: typing.Dict[int, str] = Utils.KeyedDefaultDict(lambda code: f'Unknown location (ID:{code})')
+    location_names: typing.Dict[str, typing.Dict[int, str]] = (
+        collections.defaultdict(lambda: Utils.KeyedDefaultDict(lambda code: f'Unknown location (ID:{code})')))
     location_name_groups: typing.Dict[str, typing.Dict[str, typing.Set[str]]]
     all_item_and_group_names: typing.Dict[str, typing.Set[str]]
     all_location_and_group_names: typing.Dict[str, typing.Set[str]]
@@ -267,14 +269,21 @@ def _init_game_data(self):
             if "checksum" in game_package:
                 self.checksums[game_name] = game_package["checksum"]
             for item_name, item_id in game_package["item_name_to_id"].items():
-                self.item_names[item_id] = item_name
+                self.item_names[game_name][item_id] = item_name
             for location_name, location_id in game_package["location_name_to_id"].items():
-                self.location_names[location_id] = location_name
+                self.location_names[game_name][location_id] = location_name
             self.all_item_and_group_names[game_name] = \
                 set(game_package["item_name_to_id"]) | set(self.item_name_groups[game_name])
             self.all_location_and_group_names[game_name] = \
                 set(game_package["location_name_to_id"]) | set(self.location_name_groups.get(game_name, []))
 
+        archipelago_item_names = self.item_names["Archipelago"]
+        archipelago_location_names = self.location_names["Archipelago"]
+        for game in [game_name for game_name in self.gamespackage if game_name != "Archipelago"]:
+            # Add Archipelago items and locations to each data package.
+            self.item_names[game].update(archipelago_item_names)
+            self.location_names[game].update(archipelago_location_names)
+
     def item_names_for_game(self, game: str) -> typing.Optional[typing.Dict[str, int]]:
         return self.gamespackage[game]["item_name_to_id"] if game in self.gamespackage else None
 
@@ -766,10 +775,7 @@ async def on_client_connected(ctx: Context, client: Client):
         for slot, connected_clients in clients.items():
             if connected_clients:
                 name = ctx.player_names[team, slot]
-                players.append(
-                    NetworkPlayer(team, slot,
-                                  ctx.name_aliases.get((team, slot), name), name)
-                )
+                players.append(NetworkPlayer(team, slot, ctx.name_aliases.get((team, slot), name), name))
     games = {ctx.games[x] for x in range(1, len(ctx.games) + 1)}
     games.add("Archipelago")
     await ctx.send_msgs(client, [{
@@ -784,8 +790,6 @@ async def on_client_connected(ctx: Context, client: Client):
         'permissions': get_permissions(ctx),
         'hint_cost': ctx.hint_cost,
         'location_check_points': ctx.location_check_points,
-        'datapackage_versions': {game: game_data["version"] for game, game_data
-                                 in ctx.gamespackage.items() if game in games},
         'datapackage_checksums': {game: game_data["checksum"] for game, game_data
                                   in ctx.gamespackage.items() if game in games and "checksum" in game_data},
         'seed_name': ctx.seed_name,
@@ -989,8 +993,8 @@ def register_location_checks(ctx: Context, team: int, slot: int, locations: typi
             send_items_to(ctx, team, target_player, new_item)
 
             ctx.logger.info('(Team #%d) %s sent %s to %s (%s)' % (
-                team + 1, ctx.player_names[(team, slot)], ctx.item_names[item_id],
-                ctx.player_names[(team, target_player)], ctx.location_names[location]))
+                team + 1, ctx.player_names[(team, slot)], ctx.item_names[ctx.slot_info[target_player].game][item_id],
+                ctx.player_names[(team, target_player)], ctx.location_names[ctx.slot_info[slot].game][location]))
             info_text = json_format_send_event(new_item, target_player)
             ctx.broadcast_team(team, [info_text])
 
@@ -1044,8 +1048,8 @@ def collect_hint_location_id(ctx: Context, team: int, slot: int, seeked_location
 
 def format_hint(ctx: Context, team: int, hint: NetUtils.Hint) -> str:
     text = f"[Hint]: {ctx.player_names[team, hint.receiving_player]}'s " \
-           f"{ctx.item_names[hint.item]} is " \
-           f"at {ctx.location_names[hint.location]} " \
+           f"{ctx.item_names[ctx.slot_info[hint.receiving_player].game][hint.item]} is " \
+           f"at {ctx.location_names[ctx.slot_info[hint.finding_player].game][hint.location]} " \
            f"in {ctx.player_names[team, hint.finding_player]}'s World"
 
     if hint.entrance:
@@ -1347,7 +1351,7 @@ def _cmd_remaining(self) -> bool:
         if self.ctx.remaining_mode == "enabled":
             remaining_item_ids = get_remaining(self.ctx, self.client.team, self.client.slot)
             if remaining_item_ids:
-                self.output("Remaining items: " + ", ".join(self.ctx.item_names[item_id]
+                self.output("Remaining items: " + ", ".join(self.ctx.item_names[self.client.slot.game][item_id]
                                                             for item_id in remaining_item_ids))
             else:
                 self.output("No remaining items found.")
@@ -1360,7 +1364,7 @@ def _cmd_remaining(self) -> bool:
             if self.ctx.client_game_state[self.client.team, self.client.slot] == ClientStatus.CLIENT_GOAL:
                 remaining_item_ids = get_remaining(self.ctx, self.client.team, self.client.slot)
                 if remaining_item_ids:
-                    self.output("Remaining items: " + ", ".join(self.ctx.item_names[item_id]
+                    self.output("Remaining items: " + ", ".join(self.ctx.item_names[self.client.slot.game][item_id]
                                                                 for item_id in remaining_item_ids))
                 else:
                     self.output("No remaining items found.")
@@ -1378,7 +1382,8 @@ def _cmd_missing(self, filter_text="") -> bool:
         locations = get_missing_checks(self.ctx, self.client.team, self.client.slot)
 
         if locations:
-            names = [self.ctx.location_names[location] for location in locations]
+            game = self.ctx.slot_info[self.client.slot].game
+            names = [self.ctx.location_names[game][location] for location in locations]
             if filter_text:
                 location_groups = self.ctx.location_name_groups[self.ctx.games[self.client.slot]]
                 if filter_text in location_groups:  # location group name
@@ -1403,7 +1408,8 @@ def _cmd_checked(self, filter_text="") -> bool:
         locations = get_checked_checks(self.ctx, self.client.team, self.client.slot)
 
         if locations:
-            names = [self.ctx.location_names[location] for location in locations]
+            game = self.ctx.slot_info[self.client.slot].game
+            names = [self.ctx.location_names[game][location] for location in locations]
             if filter_text:
                 location_groups = self.ctx.location_name_groups[self.ctx.games[self.client.slot]]
                 if filter_text in location_groups:  # location group name
@@ -1484,10 +1490,10 @@ def get_hints(self, input_text: str, for_location: bool = False) -> bool:
         elif input_text.isnumeric():
             game = self.ctx.games[self.client.slot]
             hint_id = int(input_text)
-            hint_name = self.ctx.item_names[hint_id] \
-                if not for_location and hint_id in self.ctx.item_names \
-                else self.ctx.location_names[hint_id] \
-                if for_location and hint_id in self.ctx.location_names \
+            hint_name = self.ctx.item_names[game][hint_id] \
+                if not for_location and hint_id in self.ctx.item_names[game] \
+                else self.ctx.location_names[game][hint_id] \
+                if for_location and hint_id in self.ctx.location_names[game] \
                 else None
             if hint_name in self.ctx.non_hintable_names[game]:
                 self.output(f"Sorry, \"{hint_name}\" is marked as non-hintable.")

NetUtils.py

@@ -247,7 +247,8 @@ def _handle_item_name(self, node: JSONMessagePart):
 
     def _handle_item_id(self, node: JSONMessagePart):
         item_id = int(node["text"])
-        node["text"] = self.ctx.item_names[item_id]
+        game = self.ctx.slot_info[node["player"]].game
+        node["text"] = self.ctx.item_names[game][item_id]
         return self._handle_item_name(node)
 
     def _handle_location_name(self, node: JSONMessagePart):
@@ -256,7 +257,8 @@ def _handle_location_name(self, node: JSONMessagePart):
 
     def _handle_location_id(self, node: JSONMessagePart):
         item_id = int(node["text"])
-        node["text"] = self.ctx.location_names[item_id]
+        game = self.ctx.slot_info[node["player"]].game
+        node["text"] = self.ctx.location_names[game][item_id]
         return self._handle_location_name(node)
 
     def _handle_entrance_name(self, node: JSONMessagePart):

UndertaleClient.py

@@ -247,8 +247,8 @@ async def process_undertale_cmd(ctx: UndertaleContext, cmd: str, args: dict):
             with open(os.path.join(ctx.save_game_folder, filename), "w") as f:
                 toDraw = ""
                 for i in range(20):
-                    if i < len(str(ctx.item_names[l.item])):
-                        toDraw += str(ctx.item_names[l.item])[i]
+                    if i < len(str(ctx.item_names[ctx.game][l.item])):
+                        toDraw += str(ctx.item_names[ctx.game][l.item])[i]
                     else:
                         break
                 f.write(toDraw)

Utils.py

@@ -46,7 +46,7 @@ def as_simple_string(self) -> str:
         return ".".join(str(item) for item in self)
 
 
-__version__ = "0.4.6"
+__version__ = "0.5.0"
 version_tuple = tuplize_version(__version__)
 
 is_linux = sys.platform.startswith("linux")
@@ -455,6 +455,9 @@ class KeyedDefaultDict(collections.defaultdict):
     """defaultdict variant that uses the missing key as argument to default_factory"""
     default_factory: typing.Callable[[typing.Any], typing.Any]
 
+    def __init__(self, default_factory: typing.Callable[[Any], Any] = None, **kwargs):
+        super().__init__(default_factory, **kwargs)
+
     def __missing__(self, key):
         self[key] = value = self.default_factory(key)
         return value

WargrooveClient.py

@@ -176,7 +176,7 @@ def on_package(self, cmd: str, args: dict):
                 if not os.path.isfile(path):
                     open(path, 'w').close()
                     # Announcing commander unlocks
-                    item_name = self.item_names[network_item.item]
+                    item_name = self.item_names[self.game][network_item.item]
                     if item_name in faction_table.keys():
                         for commander in faction_table[item_name]:
                             logger.info(f"{commander.name} has been unlocked!")
@@ -197,7 +197,7 @@ def on_package(self, cmd: str, args: dict):
                     open(print_path, 'w').close()
                     with open(print_path, 'w') as f:
                         f.write("Received " +
-                                self.item_names[network_item.item] +
+                                self.item_names[self.game][network_item.item] +
                                 " from " +
                                 self.player_names[network_item.player])
                         f.close()
@@ -342,7 +342,7 @@ def update_commander_data(self):
             faction_items = 0
             faction_item_names = [faction + ' Commanders' for faction in faction_table.keys()]
             for network_item in self.items_received:
-                if self.item_names[network_item.item] in faction_item_names:
+                if self.item_names[self.game][network_item.item] in faction_item_names:
                     faction_items += 1
             starting_groove = (faction_items - 1) * self.starting_groove_multiplier
             # Must be an integer larger than 0