Refactor code for improved type safety and null handling

Author: shima004

adf_core_python/cli/template/src/team_name/module/complex/sample_road_detector.py

@@ -37,7 +37,7 @@ def __init__(
         )
 
         self.register_sub_module(self._path_planning)
-        self._result = None
+        self._result: Optional[EntityID] = None
 
     def precompute(self, precompute_data: PrecomputeData) -> RoadDetector:
         super().precompute(precompute_data)
@@ -108,7 +108,7 @@ def update_info(self, message_manager: MessageManager) -> RoadDetector:
                 if isinstance(entity, Building):
                     self._result = None
                 elif isinstance(entity, Road):
-                    road: Road = cast(Road, entity)
+                    road = entity
                     if road.get_blockades() == []:
                         self._target_areas.remove(self._result)
                         self._result = None
@@ -117,7 +117,9 @@ def update_info(self, message_manager: MessageManager) -> RoadDetector:
 
     def calculate(self) -> RoadDetector:
         if self._result is None:
-            position_entity_id: EntityID = self._agent_info.get_position_entity_id()
+            position_entity_id = self._agent_info.get_position_entity_id()
+            if position_entity_id is None:
+                return self
             if position_entity_id in self._target_areas:
                 self._result = position_entity_id
                 return self
@@ -128,21 +130,22 @@ def calculate(self) -> RoadDetector:
 
             self._priority_roads = self._priority_roads - set(remove_list)
             if len(self._priority_roads) > 0:
-                _nearest_target_area = self._agent_info.get_position_entity_id()
+                agent_position = self._agent_info.get_position_entity_id()
+                if agent_position is None:
+                    return self
+                _nearest_target_area = agent_position
                 _nearest_distance = float("inf")
                 for target_area in self._target_areas:
                     if (
-                        self._world_info.get_distance(
-                            self._agent_info.get_position_entity_id(), target_area
-                        )
+                        self._world_info.get_distance(agent_position, target_area)
                         < _nearest_distance
                     ):
                         _nearest_target_area = target_area
                         _nearest_distance = self._world_info.get_distance(
-                            self._agent_info.get_position_entity_id(), target_area
+                            agent_position, target_area
                         )
                 path: list[EntityID] = self._path_planning.get_path(
-                    self._agent_info.get_position_entity_id(), _nearest_target_area
+                    agent_position, _nearest_target_area
                 )
                 if path is not None and len(path) > 0:
                     self._result = path[-1]

adf_core_python/cli/template/src/team_name/module/complex/sample_search.py

@@ -64,7 +64,8 @@ def update_info(self, message_manager: MessageManager) -> Search:
         )
 
         searched_building_id = self._agent_info.get_position_entity_id()
-        self._unreached_building_ids.discard(searched_building_id)
+        if searched_building_id is not None:
+            self._unreached_building_ids.discard(searched_building_id)
 
         if len(self._unreached_building_ids) == 0:
             self._unreached_building_ids = self._get_search_targets()

adf_core_python/core/agent/info/world_info.py

@@ -1,4 +1,4 @@
-from typing import Any, Optional, cast
+from typing import Any, Optional
 
 from rcrscore.entities import EntityID
 from rcrscore.entities.area import Area
@@ -199,7 +199,7 @@ def get_blockades(self, area: Area) -> set[Blockade]:
             for blockade_entity_id in blockade_entity_ids:
                 blockades_entity = self.get_entity(blockade_entity_id)
                 if isinstance(blockades_entity, Blockade):
-                    blockades.add(cast(Blockade, blockades_entity))
+                    blockades.add(blockades_entity)
         return blockades
 
     def add_entity(self, entity: Entity) -> None:

adf_core_python/core/gateway/component/module/algorithm/gateway_clustering.py

@@ -78,7 +78,9 @@ def get_cluster_entities(self, cluster_index: int) -> list[Entity]:
         entity_ids: list[int] = json.loads(json_str)
         entities: list[Entity] = []
         for entity_id in entity_ids:
-            entities.append(self._world_info.get_entity(EntityID(entity_id)))
+            entity = self._world_info.get_entity(EntityID(entity_id))
+            if entity is not None:
+                entities.append(entity)
         return entities
 
     def get_cluster_entity_ids(self, cluster_index: int) -> list[EntityID]:

adf_core_python/core/gateway/component/module/complex/gateway_target_allocator.py

@@ -63,11 +63,12 @@ def calculate(self) -> GatewayTargetAllocator:
 
     def get_result(self) -> dict[EntityID, EntityID]:
         response = self._gateway_module.execute("getResult")
-        response_keys = response.get_all_keys()
+        response_keys = response.data.keys()
         result: dict[EntityID, EntityID] = {}
         for key in response_keys:
+            value = response.get_value(key)
             result[EntityID(int(key))] = EntityID(
-                int(response.get_value_or_default(key, "-1"))
+                int(value if value is not None else "-1")
             )
 
         return result

adf_core_python/core/gateway/gateway_module.py

@@ -54,6 +54,8 @@ def initialize(self, module_name: str, default_class_name: str) -> str:
         return self.get_gateway_class_name()
 
     def get_execute_response(self) -> Config:
+        if self._result is None:
+            raise RuntimeError("No execution result available")
         return self._result
 
     def set_execute_response(self, result: Config) -> None:

adf_core_python/implement/action/default_extend_action_clear.py

@@ -59,7 +59,7 @@ def __init__(
             )
         )
 
-        self._target_entity_id = None
+        self._target_entity_id: Optional[EntityID] = None
         self._move_point_cache: dict[EntityID, Optional[set[tuple[float, float]]]] = {}
         self._old_clear_x = 0
         self._old_clear_y = 0
@@ -135,14 +135,14 @@ def calculate(self) -> ExtendAction:
             return self
 
         agent_position_entity_id = police_force.get_position()
+        if agent_position_entity_id is None:
+            return self
         target_entity = self.world_info.get_entity(self._target_entity_id)
         position_entity = self.world_info.get_entity(agent_position_entity_id)
         if target_entity is None or isinstance(target_entity, Area) is False:
             return self
         if isinstance(position_entity, Road):
-            self.result = self._get_rescue_action(
-                police_force, cast(Road, position_entity)
-            )
+            self.result = self._get_rescue_action(police_force, position_entity)
             if self.result is not None:
                 return self
 
@@ -180,7 +180,7 @@ def calculate(self) -> ExtendAction:
                         police_force, cast(Area, entity)
                     )
                     if self.result is not None and isinstance(self.result, ActionMove):
-                        action_move = cast(ActionMove, self.result)
+                        action_move = self.result
                         if action_move.is_destination_defined():
                             self.result = None
 
@@ -193,15 +193,16 @@ def _need_rest(self, police_force: PoliceForce) -> bool:
         hp = police_force.get_hp()
         damage = police_force.get_damage()
 
-        if hp == 0 or damage == 0:
+        if hp is None or damage is None or hp == 0 or damage == 0:
             return False
 
         active_time = (hp / damage) + (1 if (hp % damage) != 0 else 0)
         if self._kernel_time == -1:
             self._kernel_time = self.scenario_info.get_value("kernel.timesteps", -1)
 
-        return damage >= self._threshold_rest or (
-            active_time + self.agent_info.get_time() < self._kernel_time
+        return damage is not None and (
+            damage >= self._threshold_rest
+            or (active_time + self.agent_info.get_time() < self._kernel_time)
         )
 
     def _calc_rest(
@@ -211,6 +212,8 @@ def _calc_rest(
         target_entity_ids: list[EntityID],
     ) -> Optional[Action]:
         position_entity_id = police_force.get_position()
+        if position_entity_id is None:
+            return None
         refuges = self.world_info.get_entity_ids_of_types([Refuge])
         current_size = len(refuges)
         if position_entity_id in refuges:
@@ -244,12 +247,13 @@ def _calc_rest(
     def _get_rescue_action(
         self, police_entity: PoliceForce, road: Road
     ) -> Optional[Action]:
+        road_blockades = road.get_blockades()
         blockades = set(
             []
-            if road.get_blockades() is None
+            if road_blockades is None
             else [
                 cast(Blockade, self.world_info.get_entity(blockade_entity_id))
-                for blockade_entity_id in road.get_blockades()
+                for blockade_entity_id in road_blockades
             ]
         )
         agent_entities = set(
@@ -263,15 +267,30 @@ def _get_rescue_action(
 
         for agent_entity in agent_entities:
             human = cast(Human, agent_entity)
-            if human.get_position().get_value() != road.get_entity_id().get_value():
+            human_position = human.get_position()
+            if (
+                human_position is None
+                or human_position.get_value() != road.get_entity_id().get_value()
+            ):
                 continue
 
             human_x = human.get_x()
             human_y = human.get_y()
+            if (
+                human_x is None
+                or human_y is None
+                or police_x is None
+                or police_y is None
+            ):
+                continue
+
             action_clear: Optional[ActionClear | ActionClearArea] = None
             clear_blockade: Optional[Blockade] = None
             for blockade in blockades:
-                if not self._is_inside(human_x, human_y, blockade.get_apexes()):
+                blockade_apexes = blockade.get_apexes()
+                if blockade_apexes is None or not self._is_inside(
+                    human_x, human_y, blockade_apexes
+                ):
                     continue
 
                 distance = self._get_distance(police_x, police_y, human_x, human_y)
@@ -373,7 +392,10 @@ def _get_distance(self, x1: float, y1: float, x2: float, y2: float) -> float:
     def _is_intersecting_area(
         self, agent_x: float, agent_y: float, point_x: float, point_y: float, area: Area
     ) -> bool:
-        for edge in area.get_edges():
+        edges = area.get_edges()
+        if edges is None:
+            return False
+        for edge in edges:
             start_x = edge.get_start_x()
             start_y = edge.get_start_y()
             end_x = edge.get_end_x()
@@ -483,11 +505,13 @@ def _get_move_points(self, road: Road) -> set[tuple[float, float]]:
                     if self._is_inside(mid_x, mid_y, apex):
                         points.add((mid_x, mid_y))
 
-            for edge in road.get_edges():
-                mid_x = (edge.get_start_x() + edge.get_end_x()) / 2.0
-                mid_y = (edge.get_start_y() + edge.get_end_y()) / 2.0
-                if (mid_x, mid_y) in points:
-                    points.remove((mid_x, mid_y))
+            edges = road.get_edges()
+            if edges is not None:
+                for edge in edges:
+                    mid_x = (edge.get_start_x() + edge.get_end_x()) / 2.0
+                    mid_y = (edge.get_start_y() + edge.get_end_y()) / 2.0
+                    if (mid_x, mid_y) in points:
+                        points.remove((mid_x, mid_y))
 
             self._move_point_cache[road.get_entity_id()] = points
 
@@ -518,6 +542,8 @@ def _is_intersecting_blockade(
         blockade: Blockade,
     ) -> bool:
         apexes = blockade.get_apexes()
+        if apexes is None or len(apexes) < 4:
+            return False
         for i in range(0, len(apexes) - 3, 2):
             line1 = LineString(
                 [(apexes[i], apexes[i + 1]), (apexes[i + 2], apexes[i + 3])]
@@ -532,6 +558,8 @@ def _is_intersecting_blockades(
     ) -> bool:
         apexes1 = blockade1.get_apexes()
         apexes2 = blockade2.get_apexes()
+        if apexes1 is None or apexes2 is None or len(apexes1) < 4 or len(apexes2) < 4:
+            return False
         for i in range(0, len(apexes1) - 2, 2):
             for j in range(0, len(apexes2) - 2, 2):
                 line1 = LineString(
@@ -593,20 +621,26 @@ def _get_area_clear_action(
             if min_distance < self._clear_distance:
                 return ActionClear(clear_blockade)
             else:
-                return ActionMove(
-                    [police_entity.get_position()],
-                    clear_blockade.get_x(),
-                    clear_blockade.get_y(),
-                )
+                position = police_entity.get_position()
+                if position is not None:
+                    return ActionMove(
+                        [position],
+                        clear_blockade.get_x(),
+                        clear_blockade.get_y(),
+                    )
 
         agent_x = police_entity.get_x()
         agent_y = police_entity.get_y()
+        if agent_x is None or agent_y is None:
+            return None
         clear_blockade = None
         min_point_distance = sys.float_info.max
         clear_x = 0
         clear_y = 0
         for blockade in blockades:
             apexes = blockade.get_apexes()
+            if apexes is None or len(apexes) < 4:
+                continue
             for i in range(0, len(apexes) - 2, 2):
                 distance = self._get_distance(
                     agent_x, agent_y, apexes[i], apexes[i + 1]
@@ -625,7 +659,9 @@ def _get_area_clear_action(
                 clear_x = int(agent_x + vector[0])
                 clear_y = int(agent_y + vector[1])
                 return ActionClearArea(clear_x, clear_y)
-            return ActionMove([police_entity.get_position()], clear_x, clear_y)
+            position = police_entity.get_position()
+            if position is not None:
+                return ActionMove([position], clear_x, clear_y)
 
         return None
 
@@ -637,7 +673,12 @@ def _get_neighbour_position_action(
     ) -> Optional[Action]:
         agent_x = police_entity.get_x()
         agent_y = police_entity.get_y()
-        position = self.world_info.get_entity(police_entity.get_position())
+        if agent_x is None or agent_y is None:
+            return None
+        position_id = police_entity.get_position()
+        if position_id is None:
+            return None
+        position = self.world_info.get_entity(position_id)
         if position is None:
             return None
 
@@ -646,7 +687,7 @@ def _get_neighbour_position_action(
             return None
 
         if isinstance(position, Road):
-            road = cast(Road, position)
+            road = position
             if road.get_blockades() != []:
                 mid_x = (edge.get_start_x() + edge.get_end_x()) / 2.0
                 mid_y = (edge.get_start_y() + edge.get_end_y()) / 2.0
@@ -716,7 +757,7 @@ def _get_neighbour_position_action(
                     return action_move
 
         if isinstance(target, Road):
-            road = cast(Road, target)
+            road = target
             if road.get_blockades() == []:
                 return ActionMove([position.get_entity_id(), target.get_entity_id()])
 
@@ -726,6 +767,8 @@ def _get_neighbour_position_action(
             clear_y = 0
             for blockade in self.world_info.get_blockades(road):
                 apexes = blockade.get_apexes()
+                if apexes is None or len(apexes) < 4:
+                    continue
                 for i in range(0, len(apexes) - 2, 2):
                     distance = self._get_distance(
                         agent_x, agent_y, apexes[i], apexes[i + 1]

adf_core_python/implement/action/default_extend_action_rescue.py

@@ -117,7 +117,7 @@ def _calc_rescue(
             return None
 
         if isinstance(target_entity, Human):
-            human = cast(Human, target_entity)
+            human = target_entity
             if human.get_hp() == 0:
                 return None
 
@@ -139,7 +139,7 @@ def _calc_rescue(
             return None
 
         if isinstance(target_entity, Blockade):
-            blockade = cast(Blockade, target_entity)
+            blockade = target_entity
             blockade_position = blockade.get_position()
             if blockade_position is None:
                 return None