sc2: Adding a max supply reduction trap item; added an option to control filler ratios

worlds/sc2/__init__.py

@@ -7,7 +7,7 @@
 from Options import Accessibility
 from worlds.AutoWorld import WebWorld, World
 from .item.item_tables import (
-    filler_items, get_full_item_list, ProtossItemType,
+    get_full_item_list, ProtossItemType,
     ItemData, kerrigan_actives, kerrigan_passives,
     not_balanced_starting_units, WEAPON_ARMOR_UPGRADE_MAX_LEVEL, ZergItemType,
 )
@@ -19,8 +19,8 @@
     get_option_value, LocationInclusion, KerriganLevelItemDistribution,
     KerriganPresence, KerriganPrimalStatus, kerrigan_unit_available, StarterUnit, SpearOfAdunPresence,
     get_enabled_campaigns, SpearOfAdunAutonomouslyCastAbilityPresence, Starcraft2Options,
-    GrantStoryTech, GenericUpgradeResearch, GenericUpgradeItems, RequiredTactics,
-    upgrade_included_names, EnableVoidTrade
+    GrantStoryTech, GenericUpgradeResearch, RequiredTactics,
+    upgrade_included_names, EnableVoidTrade, FillerRatio,
 )
 from .rules import get_basic_units
 from . import settings
@@ -95,11 +95,13 @@ class SC2World(World):
     has_zerg_air_unit: bool = True
     has_protoss_ground_unit: bool = True
     has_protoss_air_unit: bool = True
+    filler_ratio: Dict[str, int]
 
     def __init__(self, multiworld: MultiWorld, player: int):
         super(SC2World, self).__init__(multiworld, player)
         self.location_cache = []
         self.locked_locations = []
+        self.filler_ratio = FillerRatio.default
 
     def create_item(self, name: str) -> Item:
         data = get_full_item_list()[name]
@@ -119,6 +121,7 @@ def create_items(self):
         # * If the item pool is less than the location count, add some filler items
 
         setup_events(self.player, self.locked_locations, self.location_cache)
+        set_up_filler_ratio(self)
 
         item_list: List[FilterItem] = create_and_flag_explicit_item_locks_and_excludes(self)
         flag_excludes_by_faction_presence(self, item_list)
@@ -166,21 +169,18 @@ def set_rules(self) -> None:
             # Forcing completed goal and minimal accessibility on no logic
             self.options.accessibility.value = Accessibility.option_minimal
             required_items = self.custom_mission_order.get_items_to_lock()
-            self.multiworld.completion_condition[self.player] = lambda state, required_items=required_items: all(
+            self.multiworld.completion_condition[self.player] = lambda state, required_items=required_items: all(  # type: ignore
                 state.has(item, self.player, amount) for (item, amount) in required_items.items()
             )
         else:
             self.multiworld.completion_condition[self.player] = self.custom_mission_order.get_completion_condition(self.player)
 
     def get_filler_item_name(self) -> str:
-        available_filler_items: list[str] = list(filler_items)
-        if self.has_protoss_ground_unit or self.has_protoss_air_unit:
-            available_filler_items.append(item_names.SHIELD_REGENERATION)
-        available_filler_items.sort()
-        return self.random.choice(available_filler_items)
-
-    def fill_slot_data(self):
-        slot_data = {}
+        # Assume `self.filler_ratio` is validated and has at least one non-zero entry
+        return self.random.choices(tuple(self.filler_ratio), weights=self.filler_ratio.values())[0]  # type: ignore
+
+    def fill_slot_data(self) -> Mapping[str, Any]:
+        slot_data: Dict[str, Any] = {}
         for option_name in [field.name for field in fields(Starcraft2Options)]:
             option = get_option_value(self, option_name)
             if type(option) in {str, int}:
@@ -232,11 +232,10 @@ def pre_fill(self) -> None:
                         item = self.multiworld.create_item(item_name, self.player)
                         self.multiworld.push_precollected(item)
 
-    def extend_hint_information(self, hint_data: Dict[int, Dict[int, str]]):
+    def extend_hint_information(self, hint_data: Dict[int, Dict[int, str]]) -> None:
         """
-        Generate information to hints where each mission is actually located in the mission order
+        Generate information to hint where each mission is actually located in the mission order
         :param hint_data:
-        :return:
         """
         hint_data[self.player] = {}
         for campaign in self.custom_mission_order.campaigns:
@@ -269,7 +268,7 @@ def extend_hint_information(self, hint_data: Dict[int, Dict[int, str]]):
                                         hint_data[self.player][location.address] = mission_position_name
 
 
-def _get_column_display(index, single_row_layout):
+def _get_column_display(index: int, single_row_layout: bool) -> str:
     """
     Helper function to display column name
     :param index:
@@ -280,11 +279,11 @@ def _get_column_display(index, single_row_layout):
         return str(index + 1)
     else:
         # Convert column name to a letter, from Z continue with AA and so on
-        f = lambda x: "" if x == 0 else f((x - 1) // 26) + chr((x - 1) % 26 + ord("A"))
+        f: Callable[[int], str] = lambda x: "" if x == 0 else f((x - 1) // 26) + chr((x - 1) % 26 + ord("A"))
         return f(index + 1)
 
 
-def setup_events(player: int, locked_locations: List[str], location_cache: List[Location]):
+def setup_events(player: int, locked_locations: List[str], location_cache: List[Location]) -> None:
     for location in location_cache:
         if location.address is None:
             item = Item(location.name, ItemClassification.progression, None, player)
@@ -741,7 +740,7 @@ def flag_and_add_resource_locations(world: SC2World, item_list: List[FilterItem]
             if (sc2_location.type in resource_location_types
                 or (sc2_location.flags & resource_location_flags)
             ):
-                item_name = world.random.choice(filler_items)
+                item_name = world.get_filler_item_name()
                 item = create_item_with_correct_settings(world.player, item_name)
                 location.place_locked_item(item)
                 world.locked_locations.append(location.name)
@@ -796,12 +795,27 @@ def item_list_contains_parent(world: SC2World, item_data: ItemData, item_name_li
     return item_parents.parent_present[item_data.parent](item_name_list, world.options)
 
 
-def pad_item_pool_with_filler(self: SC2World, num_items: int, pool: List[Item]):
+def pad_item_pool_with_filler(world: SC2World, num_items: int, pool: List[Item]):
     for _ in range(num_items):
-        item = create_item_with_correct_settings(self.player, self.get_filler_item_name())
+        item = create_item_with_correct_settings(world.player, world.get_filler_item_name())
         pool.append(item)
 
 
+def set_up_filler_ratio(world: SC2World) -> None:
+    world.filler_ratio = world.options.filler_ratio.value.copy()
+    mission_flags = world.custom_mission_order.get_used_flags()
+    include_protoss = (
+        MissionFlag.Protoss in mission_flags
+        or (world.options.take_over_ai_allies and (MissionFlag.AiProtossAlly in mission_flags))
+    )
+    if not include_protoss:
+        world.filler_ratio.pop(item_names.SHIELD_REGENERATION, 0)
+    if sum(world.filler_ratio.values()) == 0:
+        world.filler_ratio = FillerRatio.default.copy()
+    if not include_protoss:
+        world.filler_ratio.pop(item_names.SHIELD_REGENERATION, 0)
+
+
 def get_random_first_mission(world: SC2World, mission_order: SC2MissionOrder) -> SC2Mission:
     # Pick an arbitrary lowest-difficulty starer mission
     starting_missions = mission_order.get_starting_missions()
@@ -882,4 +896,4 @@ def push_precollected_items_to_multiworld(world: SC2World, item_list: List[Starc
     for item in item_list:
         if ItemFilterFlags.StartInventory not in item.filter_flags:
             continue
-        world.multiworld.push_precollected(create_item_with_correct_settings(world.player, item.name))
+        world.multiworld.push_precollected(create_item_with_correct_settings(world.player, item.name, item.filter_flags))

worlds/sc2/client.py

@@ -241,6 +241,9 @@ def _cmd_received(self, filter_search: str = "") -> bool:
         """List received items.
         Pass in a parameter to filter the search by partial item name or exact item group.
         Use '/received recent <number>' to list the last 'number' items received (default 20)."""
+        if self.ctx.slot is None:
+            self.formatted_print("Connect to a slot to view what items are received.")
+            return True
         if filter_search.casefold().startswith('recent'):
             return self._received_recent(filter_search[len('recent'):].strip())
         # Groups must be matched case-sensitively, so we properly capitalize the search term
@@ -314,6 +317,7 @@ def display_tree(
         ) -> None:
             if not should_display:
                 return
+            assert self.ctx.slot is not None
             indent_str = " " * indent
             if isinstance(element, SC2Race):
                 self.formatted_print(f" [u]{name}[/u] ")
@@ -353,7 +357,8 @@ def display_tree(
             self.formatted_print(f"[b]Filter \"{filter_search}\" found {items_obtained_matching_filter} out of {item_types_obtained} obtained item types[/b]")
         return True
 
-    def _received_recent(self, amount: str) -> None:
+    def _received_recent(self, amount: str) -> bool:
+        assert self.ctx.slot is not None
         try:
             display_amount = int(amount)
         except ValueError:
@@ -391,6 +396,9 @@ def _cmd_option(self, option_name: str = "", option_value: str = "") -> None:
             ConfigurableOptionInfo('minerals_per_item', 'minerals_per_item', options.MineralsPerItem, ConfigurableOptionType.INTEGER),
             ConfigurableOptionInfo('gas_per_item', 'vespene_per_item', options.VespenePerItem, ConfigurableOptionType.INTEGER),
             ConfigurableOptionInfo('supply_per_item', 'starting_supply_per_item', options.StartingSupplyPerItem, ConfigurableOptionType.INTEGER),
+            ConfigurableOptionInfo('max_supply_per_item', 'maximum_supply_per_item', options.MaximumSupplyPerItem, ConfigurableOptionType.INTEGER),
+            ConfigurableOptionInfo('reduced_supply_per_item', 'maximum_supply_reduction_per_item', options.MaximumSupplyReductionPerItem, ConfigurableOptionType.INTEGER),
+            ConfigurableOptionInfo('lowest_max_supply', 'lowest_maximum_supply', options.LowestMaximumSupply, ConfigurableOptionType.INTEGER),
             ConfigurableOptionInfo('no_forced_camera', 'disable_forced_camera', options.DisableForcedCamera),
             ConfigurableOptionInfo('skip_cutscenes', 'skip_cutscenes', options.SkipCutscenes),
             ConfigurableOptionInfo('enable_morphling', 'enable_morphling', options.EnableMorphling, can_break_logic=True),
@@ -622,14 +630,16 @@ def __init__(self, *args, **kwargs) -> None:
         self.vespene_per_item: int =  15  # For backwards compat with games generated pre-0.4.5
         self.starting_supply_per_item: int = 2  # For backwards compat with games generated pre-0.4.5
         self.maximum_supply_per_item: int = 2
+        self.maximum_supply_reduction_per_item: int = options.MaximumSupplyReductionPerItem.default
+        self.lowest_maximum_supply: int = options.LowestMaximumSupply.default
         self.nova_covert_ops_only = False
         self.kerrigan_levels_per_mission_completed = 0
         self.trade_enabled: int = EnableVoidTrade.default
         self.trade_underway: bool = False
         self.trade_latest_reply: typing.Optional[dict] = None
         self.trade_reply_event = asyncio.Event()
         self.trade_lock_wait: int = 0
-        self.trade_lock_start: typing.Optional[int] = None
+        self.trade_lock_start: typing.Optional[float] = None
         self.trade_response: typing.Optional[str] = None
         self.difficulty_damage_modifier: int = DifficultyDamageModifier.default
 
@@ -780,7 +790,9 @@ def on_package(self, cmd: str, args: dict) -> None:
             self.minerals_per_item = args["slot_data"].get("minerals_per_item", 15)
             self.vespene_per_item = args["slot_data"].get("vespene_per_item", 15)
             self.starting_supply_per_item = args["slot_data"].get("starting_supply_per_item", 2)
-            self.maximum_supply_per_item = args["slot_data"].get("maximum_supply_per_item", 2)
+            self.maximum_supply_per_item = args["slot_data"].get("maximum_supply_per_item", options.MaximumSupplyPerItem.default)
+            self.maximum_supply_reduction_per_item = args["slot_data"].get("maximum_supply_reduction_per_item", options.MaximumSupplyReductionPerItem.default)
+            self.lowest_maximum_supply = args["slot_data"].get("lowest_maximum_supply", options.LowestMaximumSupply.default)
             self.nova_covert_ops_only = args["slot_data"].get("nova_covert_ops_only", False)
             self.trade_enabled = args["slot_data"].get("enable_void_trade", EnableVoidTrade.option_false)
             self.difficulty_damage_modifier = args["slot_data"].get("difficulty_damage_modifier", DifficultyDamageModifier.option_true)
@@ -861,6 +873,7 @@ def parse_mission_req_table(mission_req_table: typing.Dict[SC2Campaign, typing.D
                     categories[mission.category] = []
                 mission_id = mission.mission.id
                 sub_rules: typing.List[CountMissionsRuleData] = []
+                missions: typing.List[int]
                 if mission.number:
                     amount = mission.number
                     missions = [
@@ -870,7 +883,7 @@ def parse_mission_req_table(mission_req_table: typing.Dict[SC2Campaign, typing.D
                     sub_rules.append(CountMissionsRuleData(missions, amount, [campaign_name]))
                 prev_missions: typing.List[int] = []
                 if len(mission.required_world) > 0:
-                    missions: typing.List[int] = []
+                    missions = []
                     for connection in mission.required_world:
                         if isinstance(connection, dict):
                             required_campaign = {}
@@ -1025,7 +1038,8 @@ async def trade_acquire_storage(self, keep_trying: bool = False) -> typing.Optio
                 if not keep_trying:
                     return None
                 continue
-
+
+            assert self.trade_latest_reply is not None
             reply = copy.deepcopy(self.trade_latest_reply)
 
             # Make sure the most recently received update was triggered by our lock attempt
@@ -1278,10 +1292,8 @@ def calculate_items(ctx: SC2Context) -> typing.Dict[SC2Race, typing.List[int]]:
                 accumulators[item_data.race][item_data.type.flag_word] += ctx.vespene_per_item
             elif name == item_names.STARTING_SUPPLY:
                 accumulators[item_data.race][item_data.type.flag_word] += ctx.starting_supply_per_item
-            elif name == item_names.MAX_SUPPLY:
-                accumulators[item_data.race][item_data.type.flag_word] += ctx.maximum_supply_per_item
             else:
-                accumulators[item_data.race][item_data.type.flag_word] += item_data.number
+                accumulators[item_data.race][item_data.type.flag_word] += 1
 
     # Fix Shields from generic upgrades by unit class (Maximum of ground/air upgrades)
     if shields_from_ground_upgrade > 0 or shields_from_air_upgrade > 0:
@@ -1487,6 +1499,8 @@ class ArchipelagoBot(bot.bot_ai.BotAI):
         'last_supply_used'
     ]
     ctx: SC2Context
+    # defined in bot_ai_internal.py; seems to be mis-annotated as a float and later re-annotated as an int
+    supply_used: int
 
     def __init__(self, ctx: SC2Context, mission_id: int):
         self.game_running = False
@@ -1693,7 +1707,7 @@ def get_uncollected_objectives(self) -> typing.List[int]:
             result.append(0)
         return result
 
-    def missions_beaten_count(self):
+    def missions_beaten_count(self) -> int:
         return len([location for location in self.ctx.checked_locations if location % VICTORY_MODULO == 0])
 
     async def update_colors(self):
@@ -1704,30 +1718,43 @@ async def update_colors(self):
         await self.chat_send("?SetColor nova " + str(self.ctx.player_color_nova))
         self.ctx.pending_color_update = False
 
-    async def update_resources(self, current_items):
+    async def update_resources(self, current_items: typing.Dict[SC2Race, typing.List[int]]):
+        DEFAULT_MAX_SUPPLY = 200
+        max_supply_amount = max(
+            DEFAULT_MAX_SUPPLY
+            + (
+                current_items[SC2Race.ANY][get_item_flag_word(item_names.MAX_SUPPLY)]
+                * self.ctx.maximum_supply_per_item
+            )
+            - (
+                current_items[SC2Race.ANY][get_item_flag_word(item_names.REDUCED_MAX_SUPPLY)]
+                * self.ctx.maximum_supply_reduction_per_item
+            ),
+            self.ctx.lowest_maximum_supply,
+        )
         await self.chat_send("?GiveResources {} {} {} {}".format(
             current_items[SC2Race.ANY][get_item_flag_word(item_names.STARTING_MINERALS)],
             current_items[SC2Race.ANY][get_item_flag_word(item_names.STARTING_VESPENE)],
             current_items[SC2Race.ANY][get_item_flag_word(item_names.STARTING_SUPPLY)],
-            current_items[SC2Race.ANY][get_item_flag_word(item_names.MAX_SUPPLY)]
+            max_supply_amount - DEFAULT_MAX_SUPPLY,
         ))
 
-    async def update_terran_tech(self, current_items):
+    async def update_terran_tech(self, current_items: typing.Dict[SC2Race, typing.List[int]]):
         terran_items = current_items[SC2Race.TERRAN]
         await self.chat_send("?GiveTerranTech " + " ".join(map(str, terran_items)))
 
-    async def update_zerg_tech(self, current_items, kerrigan_level):
+    async def update_zerg_tech(self, current_items: typing.Dict[SC2Race, typing.List[int]], kerrigan_level: int):
         zerg_items = current_items[SC2Race.ZERG]
         zerg_items = [value for index, value in enumerate(zerg_items) if index not in [ZergItemType.Level.flag_word, ZergItemType.Primal_Form.flag_word]]
         kerrigan_primal_by_items = kerrigan_primal(self.ctx, kerrigan_level)
         kerrigan_primal_bot_value = 1 if kerrigan_primal_by_items else 0
         await self.chat_send(f"?GiveZergTech {kerrigan_level} {kerrigan_primal_bot_value} " + ' '.join(map(str, zerg_items)))
 
-    async def update_protoss_tech(self, current_items):
+    async def update_protoss_tech(self, current_items: typing.Dict[SC2Race, typing.List[int]]):
         protoss_items = current_items[SC2Race.PROTOSS]
         await self.chat_send("?GiveProtossTech " + " ".join(map(str, protoss_items)))
 
-    async def update_misc_tech(self, current_items):
+    async def update_misc_tech(self, current_items: typing.Dict[SC2Race, typing.List[int]]):
         await self.chat_send("?GiveMiscTech {}".format(
             current_items[SC2Race.ANY][get_item_flag_word(item_names.BUILDING_CONSTRUCTION_SPEED)],
         ))

worlds/sc2/gui_config.py

@@ -59,7 +59,7 @@ def validate_color(color: Any, default: Tuple[float, float, float]) -> Tuple[Tup
         return (f'Invalid type {type(color)}; expected 3-element list or integer',), default
     elif len(color) != 3:
         return (f'Wrong number of elements in color; expected 3, got {len(color)}',), default
-    result: list[float] = [0.0, 0.0, 0.0]
+    result: List[float] = [0.0, 0.0, 0.0]
     errors: List[str] = []
     expected = 'expected a number from 0 to 1'
     for index, element in enumerate(color):
@@ -83,7 +83,7 @@ def get_button_color(race: str) -> Tuple[Tuple[str, ...], Tuple[float, float, fl
     from . import SC2World
     baseline_color = 0.345  # the button graphic is grey, with this value in each color channel
     if race == 'TERRAN':
-        user_color = SC2World.settings.terran_button_color
+        user_color: list = SC2World.settings.terran_button_color
         default_color = (0.0838, 0.2898, 0.2346)
     elif race == 'PROTOSS':
         user_color = SC2World.settings.protoss_button_color