Fix typings

worlds/civ_6/Civ6Client.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import traceback
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 import zipfile
 
 from CommonClient import ClientCommandProcessor, CommonContext, get_base_parser, logger, server_loop, gui_enabled
@@ -49,7 +49,7 @@ class CivVIContext(CommonContext):
     command_processor = CivVICommandProcessor
     game = "Civilization VI"
     items_handling = 0b111
-    tuner_sync_task = None
+    tuner_sync_task: Optional[asyncio.Task[None]] = None
     game_interface: CivVIInterface
     location_name_to_civ_location: Dict[str, CivVILocationData] = {}
     location_name_to_id: Dict[str, int] = {}
@@ -59,6 +59,7 @@ class CivVIContext(CommonContext):
     received_death_link = False
     death_link_message = ""
     death_link_enabled = False
+    slot_data: Dict[str, Any]
 
     death_link_just_changed = False
     # Used to prevent the deathlink from triggering when someone re enables it
@@ -69,21 +70,21 @@ class CivVIContext(CommonContext):
         item.name: item.code for item in generate_item_table().values()}
     connection_state = ConnectionState.DISCONNECTED
 
-    def __init__(self, server_address, password, apcivvi_file=None):
+    def __init__(self, server_address: Optional[str], password: Optional[str], apcivvi_file: Optional[str] = None):
         super().__init__(server_address, password)
-        self.slot_data = {}
+        self.slot_data: Dict[str, Any] = {}
         self.game_interface = CivVIInterface(logger)
         location_by_era = generate_era_location_table()
         self.item_table = generate_item_table()
         self.apcivvi_file = apcivvi_file
 
         # Get tables formatted in a way that is easier to use here
-        for era, locations in location_by_era.items():
-            for item_name, location in locations.items():
+        for _era, locations in location_by_era.items():
+            for _item_name, location in locations.items():
                 self.location_name_to_id[location.name] = location.code
                 self.location_name_to_civ_location[location.name] = location
 
-        for item_name, item in self.item_table.items():
+        for _item_name, item in self.item_table.items():
             self.item_id_to_civ_item[item.code] = item
 
     async def resync(self):
@@ -124,7 +125,7 @@ class CivVIManager(GameManager):
         self.ui = CivVIManager(self)
         self.ui_task = asyncio.create_task(self.ui.async_run(), name="UI")
 
-    def on_package(self, cmd: str, args: dict):
+    def on_package(self, cmd: str, args: Dict[str, Any]):
         if cmd == "Connected":
             self.slot_data = args["slot_data"]
             if "death_link" in args["slot_data"]:
@@ -133,7 +134,7 @@ def on_package(self, cmd: str, args: dict):
                     bool(args["slot_data"]["death_link"])))
 
 
-def update_connection_status(ctx: CivVIContext, status):
+def update_connection_status(ctx: CivVIContext, status: ConnectionState):
     if ctx.connection_state == status:
         return
     elif status == ConnectionState.IN_GAME:
@@ -216,7 +217,7 @@ async def handle_receive_items(ctx: CivVIContext, last_received_index_override:
             item: CivVIItemData = ctx.item_id_to_civ_item[network_item.item]
             item_to_send: CivVIItemData = ctx.item_id_to_civ_item[network_item.item]
             if index > last_received_index:
-                if item.item_type == CivVICheckType.PROGRESSIVE_DISTRICT:
+                if item.item_type == CivVICheckType.PROGRESSIVE_DISTRICT and item.civ_name:
                     # if the item is progressive, then check how far in that progression type we are and send the appropriate item
                     count = sum(
                         1 for count_item in progressive_districts if count_item.civ_name == item.civ_name)
@@ -234,7 +235,7 @@ async def handle_receive_items(ctx: CivVIContext, last_received_index_override:
                 if item.item_type == CivVICheckType.ERA:
                     count = len(progressive_eras) + 1
                     await ctx.game_interface.give_item_to_player(item_to_send, sender, count)
-                elif item.item_type == CivVICheckType.GOODY:
+                elif item.item_type == CivVICheckType.GOODY and item_to_send.civ_name:
                     item_to_send.civ_vi_id = item_to_send.civ_name
                     await ctx.game_interface.give_item_to_player(item_to_send, sender)
                 else:
@@ -280,10 +281,10 @@ async def _handle_game_ready(ctx: CivVIContext):
         await asyncio.sleep(3)
 
 
-def main(connect=None, password=None, name=None):
+def main(connect: Optional[str] = None, password: Optional[str] = None, name: Optional[str] = None):
     Utils.init_logging("Civilization VI Client")
 
-    async def _main(connect, password, name):
+    async def _main(connect: Optional[str], password: Optional[str], name: Optional[str]):
         parser = get_base_parser()
         parser.add_argument("apcivvi_file", default="", type=str, nargs="?", help="Path to apcivvi file")
         args = parser.parse_args()

worlds/civ_6/CivVIInterface.py

@@ -91,7 +91,7 @@ async def get_last_received_index(self) -> int:
         result = await self.tuner.send_game_command(command)
         return int(result)
 
-    async def send_notification(self, item: CivVIItemData, sender="someone") -> None:
+    async def send_notification(self, item: CivVIItemData, sender: str = "someone") -> None:
         command = f"GameCore.NotificationManager:SendNotification(GameCore.NotificationTypes.USER_DEFINED_2, \"{item.name} Received\", \"You have received {item.name} from \" .. \"{sender}\", 0, {item.civ_vi_id})"
         await self.tuner.send_command(command)
 

worlds/civ_6/Container.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 import os
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING, Dict, List, Optional
 import zipfile
 from BaseClasses import ItemClassification, Location
 from worlds.Files import APContainer
@@ -30,10 +30,10 @@ class CivVIContainer(APContainer):
     """
     Responsible for generating the dynamic mod files for the Civ VI multiworld
     """
-    game: str = "Civilization VI"
+    game: Optional[str] = "Civilization VI"
 
-    def __init__(self, patch_data: dict, base_path: str, output_directory: str,
-                 player=None, player_name: str = "", server: str = ""):
+    def __init__(self, patch_data: Dict[str, str], base_path: str, output_directory: str,
+                 player: Optional[int] = None, player_name: str = "", server: str = ""):
         self.patch_data = patch_data
         self.file_path = base_path
         container_path = os.path.join(output_directory, base_path + ".apcivvi")
@@ -45,7 +45,7 @@ def write_contents(self, opened_zipfile: zipfile.ZipFile) -> None:
         super().write_contents(opened_zipfile)
 
 
-def get_cost(world, location: CivVILocationData) -> int:
+def get_cost(world: 'CivVIWorld', location: CivVILocationData) -> int:
     """
     Returns the cost of the item based on the game options
     """
@@ -55,7 +55,7 @@ def get_cost(world, location: CivVILocationData) -> int:
     return int(world.location_table[location.name].cost * multiplier)
 
 
-def get_formatted_player_name(world, player) -> str:
+def get_formatted_player_name(world: 'CivVIWorld', player: int) -> str:
     """
     Returns the name of the player in the world
     """
@@ -66,7 +66,7 @@ def get_formatted_player_name(world, player) -> str:
 
 
 def get_advisor_type(world: 'CivVIWorld', location: Location) -> str:
-    if world.options.advisor_show_progression_items and location.item.classification == ItemClassification.progression:
+    if world.options.advisor_show_progression_items and location.item and location.item.classification == ItemClassification.progression:
         return "ADVISOR_PROGRESSIVE"
     else:
         return "ADVISOR_GENERIC"
@@ -76,7 +76,8 @@ def generate_new_items(world: 'CivVIWorld') -> str:
     """
     Generates the XML for the new techs/civics as well as the blockers used to prevent players from researching their own items
     """
-    locations: List[CivVILocation] = world.multiworld.get_filled_locations(world.player)
+    raw_locations = world.multiworld.get_filled_locations(world.player)
+    locations: List[CivVILocation] = [location for location in raw_locations if isinstance(location, CivVILocation)]
     techs = [location for location in locations if location.location_type ==
              CivVICheckType.TECH]
     civics = [location for location in locations if location.location_type ==
@@ -118,7 +119,7 @@ def generate_new_items(world: 'CivVIWorld') -> str:
                 f'Description="{location.name}" '
                 f'AdvisorType="{get_advisor_type(world, location)}"'
                 f'/>{nl}'
-                for location in techs])}
+                for location in techs if location.item])}
   </Technologies>
   <TechnologyPrereqs>
   {"".join([f'{tab}<Row Technology="{location.name}" PrereqTech="TECH_BLOCKER" />{nl}' for location in boost_techs])}
@@ -134,7 +135,7 @@ def generate_new_items(world: 'CivVIWorld') -> str:
                     f'Description="{location.name}" '
                     f'AdvisorType="{get_advisor_type(world, location)}"'
                     f'/>{nl}'
-                    for location in civics])}
+                    for location in civics if location.item])}
   </Civics>
   <CivicPrereqs>
   {"".join([f'{tab}<Row Civic="{location.name}" PrereqCivic="CIVIC_BLOCKER" />{nl}' for location in boost_civics])}
@@ -152,7 +153,7 @@ def generate_new_items(world: 'CivVIWorld') -> str:
     """
 
 
-def generate_setup_file(world) -> str:
+def generate_setup_file(world: 'CivVIWorld') -> str:
     """
     Generates the Lua for the setup file. This sets initial variables and state that affect gameplay around Progressive Eras
     """
@@ -177,7 +178,7 @@ def generate_setup_file(world) -> str:
     return setup
 
 
-def generate_goody_hut_sql(world) -> str:
+def generate_goody_hut_sql(world: 'CivVIWorld') -> str:
     """
     Generates the SQL for the goody huts or an empty string if they are disabled since the mod expects the file to be there
     """
@@ -213,7 +214,7 @@ def generate_goody_hut_sql(world) -> str:
         return "-- Goody Huts are disabled, no changes needed"
 
 
-def generate_update_boosts_sql(world) -> str:
+def generate_update_boosts_sql(world: 'CivVIWorld') -> str:
     """
     Generates the SQL for existing boosts in boostsanity or an empty string if they are disabled since the mod expects the file to be there
     """

worlds/civ_6/Data.py

@@ -2,18 +2,20 @@
 import json
 import os
 import pkgutil
-from typing import Dict, List
+from typing import Any, Dict, List, TypedDict
 
 
-_cache = {}
+_cache: Dict[Any, Any] = {}
 
 
-def _get_data(key: str):
+def _get_data(key: str) -> Any:
     global _cache
     if key not in _cache:
         path = os.path.join("data", f"{key}.json")
-        _cache[key] = json.loads(
-            pkgutil.get_data(__name__, path).decode())
+        data = pkgutil.get_data(__name__, path)
+        if data is None:
+            raise FileNotFoundError(f"Data file not found: {path}")
+        _cache[key] = json.loads(data.decode())
     return _cache[key]
 
 
@@ -28,7 +30,7 @@ class CivVIBoostData:
 
 def get_boosts_data() -> List[CivVIBoostData]:
     boosts_json = _get_data("boosts")
-    boosts = []
+    boosts: List[CivVIBoostData] = []
     for boost in boosts_json:
         boosts.append(CivVIBoostData(
             Type=boost["Type"],
@@ -45,33 +47,60 @@ def get_era_required_items_data() -> Dict[str, List[str]]:
     return _get_data("era_required_items")
 
 
-def get_existing_civics_data():
+class NewItemData(TypedDict):
+    Type: str
+    Cost: int
+    UITreeRow: int
+    EraType: str
+
+
+class ExistingItemData(NewItemData):
+    Name: str
+
+
+def get_existing_civics_data() -> List[ExistingItemData]:
     return _get_data("existing_civics")
 
 
-def get_existing_techs_data():
+def get_existing_techs_data() -> List[ExistingItemData]:
     return _get_data("existing_tech")
 
 
-def get_goody_hut_rewards_data():
+class GoodyHutRewardData(TypedDict):
+    Type: str
+    Name: str
+    Rarity: str
+
+
+def get_goody_hut_rewards_data() -> List[GoodyHutRewardData]:
     return _get_data("goody_hut_rewards")
 
 
-def get_new_civic_prereqs_data():
+class CivicPrereqData(TypedDict):
+    Civic: str
+    PrereqTech: str
+
+
+def get_new_civic_prereqs_data() -> List[CivicPrereqData]:
     return _get_data("new_civic_prereqs")
 
 
-def get_new_civics_data():
+def get_new_civics_data() -> List[NewItemData]:
     return _get_data("new_civics")
 
 
-def get_new_tech_prereqs_data():
+class TechPrereqData(TypedDict):
+    Technology: str
+    PrereqTech: str
+
+
+def get_new_tech_prereqs_data() -> List[TechPrereqData]:
     return _get_data("new_tech_prereqs")
 
 
-def get_new_techs_data():
+def get_new_techs_data() -> List[NewItemData]:
     return _get_data("new_tech")
 
 
-def get_progressive_districts_data():
+def get_progressive_districts_data() -> Dict[str, List[str]]:
     return _get_data("progressive_districts")

worlds/civ_6/DeathLink.py

@@ -8,7 +8,7 @@
 DEATH_LINK_EFFECTS = ["Gold", "Faith", "Era Score", "Unit Killed"]
 
 
-async def handle_receive_deathlink(ctx: 'CivVIContext', message):
+async def handle_receive_deathlink(ctx: 'CivVIContext', message: str):
     """Resolves the effects of a deathlink received from the multiworld based on the options selected by the player"""
     chosen_effect = ctx.slot_data["death_link_effect"]
     effect: str = "Gold"
@@ -73,6 +73,8 @@ async def handle_check_deathlink(ctx: 'CivVIContext'):
                     f"is reconsidering their pursuit of a domination victory",
                     f"had their plans toppled by a {result}",
                     ]
-        player = ctx.player_names[ctx.slot]
-        message = random.choice(messages)
-        await ctx.send_death(f"{player} {message}")
+
+        if ctx.slot is not None:
+            player = ctx.player_names[ctx.slot]
+            message = random.choice(messages)
+            await ctx.send_death(f"{player} {message}")