Core: Plando Items "rewrite"

BaseClasses.py

@@ -83,6 +83,8 @@ class MultiWorld():
     start_location_hints: Dict[int, Options.StartLocationHints]
     item_links: Dict[int, Options.ItemLinks]
 
+    plando_item_blocks: Dict[int, Any]
+
     game: Dict[int, str]
 
     random: random.Random
@@ -160,13 +162,15 @@ def __init__(self, players: int):
         self.local_early_items = {player: {} for player in self.player_ids}
         self.indirect_connections = {}
         self.start_inventory_from_pool: Dict[int, Options.StartInventoryPool] = {}
+        self.plando_item_blocks = {}
 
         for player in range(1, players + 1):
             def set_player_attr(attr: str, val) -> None:
                 self.__dict__.setdefault(attr, {})[player] = val
             set_player_attr('plando_items', [])
             set_player_attr('plando_texts', {})
             set_player_attr('plando_connections', [])
+            set_player_attr('plando_item_blocks', [])
             set_player_attr('game', "Archipelago")
             set_player_attr('completion_condition', lambda state: True)
         self.worlds = {}
@@ -427,7 +431,7 @@ def get_entrance(self, entrance_name: str, player: int) -> Entrance:
     def get_location(self, location_name: str, player: int) -> Location:
         return self.regions.location_cache[player][location_name]
 
-    def get_all_state(self, use_cache: bool, allow_partial_entrances: bool = False) -> CollectionState:
+    def get_all_state(self, use_cache: bool, collect_pre_fill_items: bool = True, allow_partial_entrances: bool = False) -> CollectionState:
         cached = getattr(self, "_all_state", None)
         if use_cache and cached:
             return cached.copy()
@@ -436,10 +440,11 @@ def get_all_state(self, use_cache: bool, allow_partial_entrances: bool = False)
 
         for item in self.itempool:
             self.worlds[item.player].collect(ret, item)
-        for player in self.player_ids:
-            subworld = self.worlds[player]
-            for item in subworld.get_pre_fill_items():
-                subworld.collect(ret, item)
+        if collect_pre_fill_items:
+            for player in self.player_ids:
+                subworld = self.worlds[player]
+                for item in subworld.get_pre_fill_items():
+                    subworld.collect(ret, item)
         ret.sweep_for_advancements()
 
         if use_cache:

Generate.py

@@ -298,12 +298,6 @@ def handle_name(name: str, player: int, name_counter: Counter):
     return new_name
 
 
-def roll_percentage(percentage: Union[int, float]) -> bool:
-    """Roll a percentage chance.
-    percentage is expected to be in range [0, 100]"""
-    return random.random() < (float(percentage) / 100)
-
-
 def update_weights(weights: dict, new_weights: dict, update_type: str, name: str) -> dict:
     logging.debug(f'Applying {new_weights}')
     cleaned_weights = {}
@@ -369,7 +363,7 @@ def roll_linked_options(weights: dict) -> dict:
         if "name" not in option_set:
             raise ValueError("One of your linked options does not have a name.")
         try:
-            if roll_percentage(option_set["percentage"]):
+            if Options.roll_percentage(option_set["percentage"]):
                 logging.debug(f"Linked option {option_set['name']} triggered.")
                 new_options = option_set["options"]
                 for category_name, category_options in new_options.items():
@@ -402,7 +396,7 @@ def roll_triggers(weights: dict, triggers: list, valid_keys: set) -> dict:
             trigger_result = get_choice("option_result", option_set)
             result = get_choice(key, currently_targeted_weights)
             currently_targeted_weights[key] = result
-            if result == trigger_result and roll_percentage(get_choice("percentage", option_set, 100)):
+            if result == trigger_result and Options.roll_percentage(get_choice("percentage", option_set, 100)):
                 for category_name, category_options in option_set["options"].items():
                     currently_targeted_weights = weights
                     if category_name:
@@ -501,8 +495,6 @@ def roll_settings(weights: dict, plando_options: PlandoOptions = PlandoOptions.b
         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 = copy.deepcopy(game_weights.get("plando_items", []))
     if ret.game == "A Link to the Past":
         roll_alttp_settings(ret, game_weights)
 

Main.py

@@ -11,8 +11,8 @@
 
 import worlds
 from BaseClasses import CollectionState, Item, Location, LocationProgressType, MultiWorld, Region
-from Fill import FillError, balance_multiworld_progression, distribute_items_restrictive, distribute_planned, \
-    flood_items
+from Fill import FillError, balance_multiworld_progression, distribute_items_restrictive, flood_items, \
+    parse_planned_blocks, distribute_planned_blocks, resolve_early_locations_for_planned
 from Options import StartInventoryPool
 from Utils import __version__, output_path, version_tuple, get_settings
 from settings import get_settings
@@ -148,7 +148,9 @@ def main(args, seed=None, baked_server_options: Optional[Dict[str, object]] = No
     else:
         multiworld.worlds[1].options.non_local_items.value = set()
         multiworld.worlds[1].options.local_items.value = set()
-
+
+    multiworld.plando_item_blocks = parse_planned_blocks(multiworld)
+
     AutoWorld.call_all(multiworld, "generate_basic")
 
     # remove starting inventory from pool items.
@@ -192,8 +194,9 @@ def main(args, seed=None, baked_server_options: Optional[Dict[str, object]] = No
         multiworld._all_state = None
 
     logger.info("Running Item Plando.")
-
-    distribute_planned(multiworld)
+    resolve_early_locations_for_planned(multiworld)
+    distribute_planned_blocks(multiworld, [x for player in multiworld.plando_item_blocks
+                                           for x in multiworld.plando_item_blocks[player]])
 
     logger.info('Running Pre Main Fill.')
 

Options.py

@@ -23,6 +23,12 @@
     import pathlib
 
 
+def roll_percentage(percentage: int | float) -> bool:
+    """Roll a percentage chance.
+    percentage is expected to be in range [0, 100]"""
+    return random.random() < (float(percentage) / 100)
+
+
 class OptionError(ValueError):
     pass
 
@@ -976,7 +982,7 @@ def from_any(cls, data: PlandoTextsFromAnyType) -> Self:
         if isinstance(data, typing.Iterable):
             for text in data:
                 if isinstance(text, typing.Mapping):
-                    if random.random() < float(text.get("percentage", 100)/100):
+                    if roll_percentage(text.get("percentage", 100)):
                         at = text.get("at", None)
                         if at is not None:
                             if isinstance(at, dict):
@@ -1002,7 +1008,7 @@ def from_any(cls, data: PlandoTextsFromAnyType) -> Self:
                         else:
                             raise OptionError("\"at\" must be a valid string or weighted list of strings!")
                 elif isinstance(text, PlandoText):
-                    if random.random() < float(text.percentage/100):
+                    if roll_percentage(text.percentage):
                         texts.append(text)
                 else:
                     raise Exception(f"Cannot create plando text from non-dictionary type, got {type(text)}")
@@ -1126,7 +1132,7 @@ def from_any(cls, data: PlandoConFromAnyType) -> Self:
         for connection in data:
             if isinstance(connection, typing.Mapping):
                 percentage = connection.get("percentage", 100)
-                if random.random() < float(percentage / 100):
+                if roll_percentage(percentage):
                     entrance = connection.get("entrance", None)
                     if is_iterable_except_str(entrance):
                         entrance = random.choice(sorted(entrance))
@@ -1144,7 +1150,7 @@ def from_any(cls, data: PlandoConFromAnyType) -> Self:
                         percentage
                     ))
             elif isinstance(connection, PlandoConnection):
-                if random.random() < float(connection.percentage / 100):
+                if roll_percentage(connection.percentage):
                     value.append(connection)
             else:
                 raise Exception(f"Cannot create connection from non-Dict type, got {type(connection)}.")
@@ -1420,6 +1426,116 @@ def verify(self, world: typing.Type[World], player_name: str, plando_options: "P
             link["item_pool"] = list(pool)
 
 
+@dataclass(frozen=True)
+class PlandoItem:
+    items: list[str] | dict[str, typing.Any]
+    locations: list[str]
+    world: int | str | bool | None | typing.Iterable[str] | set[int] = False
+    from_pool: bool = True
+    force: bool | typing.Literal["silent"] = "silent"
+    count: int | bool | dict[str, int] = False
+    percentage: int = 100
+
+
+class PlandoItems(Option[typing.List[PlandoItem]]):
+    """Generic items plando."""
+    default = ()
+    supports_weighting = False
+    display_name = "Plando Items"
+
+    def __init__(self, value: typing.Iterable[PlandoItem]) -> None:
+        self.value = list(deepcopy(value))
+        super().__init__()
+
+    @classmethod
+    def from_any(cls, data: typing.Any) -> Option[typing.List[PlandoItem]]:
+        if not isinstance(data, typing.Iterable):
+            raise Exception(f"Cannot create plando items from non-Iterable type, got {type(data)}")
+
+        value: typing.List[PlandoItem] = []
+        for item in data:
+            if isinstance(item, typing.Mapping):
+                percentage = item.get("percentage", 100)
+                if not isinstance(percentage, int):
+                    raise Exception(f"Plando `percentage` has to be int, not {type(percentage)}.")
+                if not (0 <= percentage <= 100):
+                    raise Exception(f"Plando `percentage` has to be between 0 and 100 (inclusive) not {percentage}.")
+                if roll_percentage(percentage):
+                    count = item.get("count", False)
+                    items = item.get("items", [])
+                    if not items:
+                        items = item.get("item", None)  # explicitly throw an error here if not present
+                        if not items:
+                            raise Exception("You must specify at least one item to place items with plando.")
+                        count = 1
+                        if isinstance(items, str):
+                            items = [items]
+                        elif not isinstance(items, dict):
+                            raise Exception(f"Plando 'item' has to be string or dictionary, not {type(items)}.")
+                    locations = item.get("locations", [])
+                    if not locations:
+                        locations = item.get("location", [])
+                        if locations:
+                            count = 1
+                        if isinstance(locations, str):
+                            locations = [locations]
+                        if not isinstance(locations, list):
+                            raise Exception(f"Plando `location` has to be string or list, not {type(locations)}")
+                    world = item.get("world", False)
+                    from_pool = item.get("from_pool", True)
+                    force = item.get("force", "silent")
+                    value.append(PlandoItem(items, locations, world, from_pool, force, count, percentage))
+            elif isinstance(item, PlandoItem):
+                if roll_percentage(item.percentage):
+                    value.append(item)
+            else:
+                raise Exception(f"Cannot create plando item from non-Dict type, got {type(item)}.")
+        return cls(value)
+
+    def verify(self, world: typing.Type[World], player_name: str, plando_options: "PlandoOptions") -> None:
+        if not self.value:
+            return
+        from BaseClasses import PlandoOptions
+        if not (PlandoOptions.items & plando_options):
+            # plando is disabled but plando options were given so overwrite the options
+            self.value = []
+            logging.warning(f"The plando items module is turned off, "
+                            f"so items for {player_name} will be ignored.")
+        else:
+            # filter down item groups
+            for plando in self.value:
+                items_copy = plando.items.copy()
+                if isinstance(plando.items, dict):
+                    for item in items_copy:
+                        if item in world.item_name_groups:
+                            value = plando.items.pop(item)
+                            group = sorted(world.item_name_groups[item])
+                            for group_item in group:
+                                if group_item in plando.items:
+                                    raise Exception(f"Plando `items` contains both \"{group_item}\" and the group "
+                                                    f"\"{item}\" which contains it. It cannot have both.")
+                            plando.items.update({key: value for key in group})
+                else:
+                    assert isinstance(plando.items, list)  # pycharm can't figure out the hinting without the hint
+                    for item in items_copy:
+                        if item in world.item_name_groups:
+                            plando.items.remove(item)
+                            plando.items.extend(sorted(world.item_name_groups[item]))
+
+    @classmethod
+    def get_option_name(cls, value: typing.List[PlandoItem]) -> str:
+        return ", ".join(["%s-%s" % (item.items, item.locations) for item in value])  #TODO: see what a better way to display would be
+
+    def __getitem__(self, index: typing.SupportsIndex) -> PlandoItem:
+        return self.value.__getitem__(index)
+
+    def __iter__(self) -> typing.Iterator[PlandoItem]:
+        yield from self.value
+
+    def __len__(self) -> int:
+        return len(self.value)
+
+
 class Removed(FreeText):
     """This Option has been Removed."""
     rich_text_doc = True
@@ -1442,6 +1558,7 @@ class PerGameCommonOptions(CommonOptions):
     exclude_locations: ExcludeLocations
     priority_locations: PriorityLocations
     item_links: ItemLinks
+    plando_items: PlandoItems
 
 
 @dataclass

entrance_rando.py

@@ -151,7 +151,7 @@ def __init__(self, world: World, coupled: bool):
         self.pairings = []
         self.world = world
         self.coupled = coupled
-        self.collection_state = world.multiworld.get_all_state(False, True)
+        self.collection_state = world.multiworld.get_all_state(False, True, True)
 
     @property
     def placed_regions(self) -> set[Region]:

test/general/test_implemented.py

@@ -53,6 +53,22 @@ def test_no_failed_world_loads(self):
         if failed_world_loads:
             self.fail(f"The following worlds failed to load: {failed_world_loads}")
 
+    def test_prefill_items(self):
+        """Test that every world can reach every location from allstate before pre_fill."""
+        for gamename, world_type in AutoWorldRegister.world_types.items():
+            if gamename not in ("Archipelago", "Sudoku", "Final Fantasy", "Test Game"):
+                with self.subTest(gamename):
+                    multiworld = setup_solo_multiworld(world_type, ("generate_early", "create_regions", "create_items",
+                                                                    "set_rules", "generate_basic"))
+                    allstate = multiworld.get_all_state(False)
+                    locations = multiworld.get_locations()
+                    reachable = multiworld.get_reachable_locations(allstate)
+                    unreachable = [location for location in locations if location not in reachable]
+
+                    self.assertTrue(not unreachable,
+                                    f"Locations were not reachable with all state before prefill: "
+                                    f"{unreachable}")
+
     def test_explicit_indirect_conditions_spheres(self):
         """Tests that worlds using explicit indirect conditions produce identical spheres as when using implicit
         indirect conditions"""