Merge branch 'main' into black-sliver-update-cx_freeze

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.lookup_in_slot(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
@@ -8,6 +9,7 @@
 import typing
 import time
 import functools
+import warnings
 
 import ModuleUpdate
 ModuleUpdate.update()
@@ -173,10 +175,74 @@ class CommonContext:
     items_handling: typing.Optional[int] = None
     want_slot_data: bool = True  # should slot_data be retrieved via Connect
 
-    # 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})')
+    class NameLookupDict:
+        """A specialized dict, with helper methods, for id -> name item/location data package lookups by game."""
+        def __init__(self, ctx: CommonContext, lookup_type: typing.Literal["item", "location"]):
+            self.ctx: CommonContext = ctx
+            self.lookup_type: typing.Literal["item", "location"] = lookup_type
+            self._unknown_item: typing.Callable[[int], str] = lambda key: f"Unknown {lookup_type} (ID: {key})"
+            self._archipelago_lookup: typing.Dict[int, str] = {}
+            self._flat_store: typing.Dict[int, str] = Utils.KeyedDefaultDict(self._unknown_item)
+            self._game_store: typing.Dict[str, typing.ChainMap[int, str]] = collections.defaultdict(
+                lambda: collections.ChainMap(self._archipelago_lookup, Utils.KeyedDefaultDict(self._unknown_item)))
+            self.warned: bool = False
+
+        # noinspection PyTypeChecker
+        def __getitem__(self, key: str) -> typing.Mapping[int, str]:
+            # TODO: In a future version (0.6.0?) this should be simplified by removing implicit id lookups support.
+            if isinstance(key, int):
+                if not self.warned:
+                    # Use warnings instead of logger to avoid deprecation message from appearing on user side.
+                    self.warned = True
+                    warnings.warn(f"Implicit name lookup by id only is deprecated and only supported to maintain "
+                                  f"backwards compatibility for now. If multiple games share the same id for a "
+                                  f"{self.lookup_type}, name could be incorrect. Please use "
+                                  f"`{self.lookup_type}_names.lookup_in_game()` or "
+                                  f"`{self.lookup_type}_names.lookup_in_slot()` instead.")
+                return self._flat_store[key]  # type: ignore
+
+            return self._game_store[key]
+
+        def __len__(self) -> int:
+            return len(self._game_store)
+
+        def __iter__(self) -> typing.Iterator[str]:
+            return iter(self._game_store)
+
+        def __repr__(self) -> str:
+            return self._game_store.__repr__()
+
+        def lookup_in_game(self, code: int, game_name: typing.Optional[str] = None) -> str:
+            """Returns the name for an item/location id in the context of a specific game or own game if `game` is
+            omitted.
+            """
+            if game_name is None:
+                game_name = self.ctx.game
+                assert game_name is not None, f"Attempted to lookup {self.lookup_type} with no game name available."
+
+            return self._game_store[game_name][code]
+
+        def lookup_in_slot(self, code: int, slot: typing.Optional[int] = None) -> str:
+            """Returns the name for an item/location id in the context of a specific slot or own slot if `slot` is
+            omitted.
+            """
+            if slot is None:
+                slot = self.ctx.slot
+                assert slot is not None, f"Attempted to lookup {self.lookup_type} with no slot info available."
+
+            return self.lookup_in_game(code, self.ctx.slot_info[slot].game)
+
+        def update_game(self, game: str, name_to_id_lookup_table: typing.Dict[str, int]) -> None:
+            """Overrides existing lookup tables for a particular game."""
+            id_to_name_lookup_table = Utils.KeyedDefaultDict(self._unknown_item)
+            id_to_name_lookup_table.update({code: name for name, code in name_to_id_lookup_table.items()})
+            self._game_store[game] = collections.ChainMap(self._archipelago_lookup, id_to_name_lookup_table)
+            self._flat_store.update(id_to_name_lookup_table)  # Only needed for legacy lookup method.
+            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(id_to_name_lookup_table)
 
     # defaults
     starting_reconnect_delay: int = 5
@@ -231,7 +297,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 +337,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.NameLookupDict(self, "item")
+        self.location_names = self.NameLookupDict(self, "location")
+
         self.jsontotextparser = JSONtoTextParser(self)
         self.rawjsontotextparser = RawJSONtoTextParser(self)
         self.update_data_package(network_data_package)
@@ -486,19 +555,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)

Generate.py

@@ -23,9 +23,7 @@
 from settings import get_settings
 from Utils import parse_yamls, version_tuple, __version__, tuplize_version
 from worlds.alttp.EntranceRandomizer import parse_arguments
-from worlds.alttp.Text import TextTable
 from worlds.AutoWorld import AutoWorldRegister
-from worlds.generic import PlandoConnection
 from worlds import failed_world_loads
 
 
@@ -432,7 +430,6 @@ def handle_option(ret: argparse.Namespace, game_weights: dict, option_key: str,
                 player_option = option.from_any(game_weights[option_key])
             else:
                 player_option = option.from_any(get_choice(option_key, game_weights))
-            del game_weights[option_key]
         else:
             player_option = option.from_any(option.default)  # call the from_any here to support default "random"
         setattr(ret, option_key, player_option)
@@ -446,9 +443,9 @@ def roll_settings(weights: dict, plando_options: PlandoOptions = PlandoOptions.b
     if "linked_options" in weights:
         weights = roll_linked_options(weights)
 
-    valid_trigger_names = set()
+    valid_keys = set()
     if "triggers" in weights:
-        weights = roll_triggers(weights, weights["triggers"], valid_trigger_names)
+        weights = roll_triggers(weights, weights["triggers"], valid_keys)
 
     requirements = weights.get("requires", {})
     if requirements:
@@ -490,7 +487,7 @@ def roll_settings(weights: dict, plando_options: PlandoOptions = PlandoOptions.b
             raise Exception(f"Remove tag cannot be used outside of trigger contexts. Found {weight}")
 
     if "triggers" in game_weights:
-        weights = roll_triggers(weights, game_weights["triggers"], valid_trigger_names)
+        weights = roll_triggers(weights, game_weights["triggers"], valid_keys)
         game_weights = weights[ret.game]
 
     ret.name = get_choice('name', weights)
@@ -499,42 +496,20 @@ def roll_settings(weights: dict, plando_options: PlandoOptions = PlandoOptions.b
 
     for option_key, option in world_type.options_dataclass.type_hints.items():
         handle_option(ret, game_weights, option_key, option, plando_options)
+        valid_keys.add(option_key)
     for option_key in game_weights:
-        if option_key in {"triggers", *valid_trigger_names}:
+        if option_key in {"triggers", *valid_keys}:
             continue
         logging.warning(f"{option_key} is not a valid option name for {ret.game} and is not present in triggers.")
     if PlandoOptions.items in plando_options:
         ret.plando_items = game_weights.get("plando_items", [])
     if ret.game == "A Link to the Past":
-        roll_alttp_settings(ret, game_weights, plando_options)
-    if PlandoOptions.connections in plando_options:
-        ret.plando_connections = []
-        options = game_weights.get("plando_connections", [])
-        for placement in options:
-            if roll_percentage(get_choice("percentage", placement, 100)):
-                ret.plando_connections.append(PlandoConnection(
-                    get_choice("entrance", placement),
-                    get_choice("exit", placement),
-                    get_choice("direction", placement, "both")
-                ))
+        roll_alttp_settings(ret, game_weights)
 
     return ret
 
 
-def roll_alttp_settings(ret: argparse.Namespace, weights, plando_options):
-
-    ret.plando_texts = {}
-    if PlandoOptions.texts in plando_options:
-        tt = TextTable()
-        tt.removeUnwantedText()
-        options = weights.get("plando_texts", [])
-        for placement in options:
-            if roll_percentage(get_choice_legacy("percentage", placement, 100)):
-                at = str(get_choice_legacy("at", placement))
-                if at not in tt:
-                    raise Exception(f"No text target \"{at}\" found.")
-                ret.plando_texts[at] = str(get_choice_legacy("text", placement))
-
+def roll_alttp_settings(ret: argparse.Namespace, weights):
     ret.sprite_pool = weights.get('sprite_pool', [])
     ret.sprite = get_choice_legacy('sprite', weights, "Link")
     if 'random_sprite_on_event' in weights:

Main.py

@@ -372,6 +372,17 @@ def precollect_hint(location):
 
                 checks_in_area: Dict[int, Dict[str, Union[int, List[int]]]] = {}
 
+                # get spheres -> filter address==None -> skip empty
+                spheres: List[Dict[int, Set[int]]] = []
+                for sphere in multiworld.get_spheres():
+                    current_sphere: Dict[int, Set[int]] = collections.defaultdict(set)
+                    for sphere_location in sphere:
+                        if type(sphere_location.address) is int:
+                            current_sphere[sphere_location.player].add(sphere_location.address)
+
+                    if current_sphere:
+                        spheres.append(dict(current_sphere))
+
                 multidata = {
                     "slot_data": slot_data,
                     "slot_info": slot_info,
@@ -386,6 +397,7 @@ def precollect_hint(location):
                     "tags": ["AP"],
                     "minimum_versions": minimum_versions,
                     "seed_name": multiworld.seed_name,
+                    "spheres": spheres,
                     "datapackage": data_package,
                 }
                 AutoWorld.call_all(multiworld, "modify_multidata", multidata)