Merge pull request #51 from adf-python/adf/impl/module/complex_49

adf/impl/module/complex/のAmbulanceで使われている部分の実装

Author: shima004

adf_core_python/core/agent/info/scenario_info.py

@@ -1,4 +1,5 @@
 from enum import Enum
+from typing import Any
 
 from adf_core_python.core.config.config import Config
 
@@ -57,18 +58,20 @@ def get_mode(self) -> Mode:
         """
         return self._mode
 
-    def get_config_value(self, key: str, default: str) -> str:
+    def get_value(self, key: str, default: Any) -> Any:
         """
         Get the value of the configuration
 
         Parameters
         ----------
         key : str
             Key of the configuration
+        default : Any
+            Default value of the configuration
 
         Returns
         -------
-        str
+        Any
             Value of the configuration
         """
         return self._config.get_value(key, default)

adf_core_python/core/agent/info/world_info.py

@@ -53,14 +53,16 @@ def get_entity(self, entity_id: EntityID) -> Optional[Entity]:
         """
         return self._world_model.get_entity(entity_id)
 
-    def get_entity_ids_of_type(self, entity_type: type[Entity]) -> list[EntityID]:
+    def get_entity_ids_of_types(
+        self, entity_types: list[type[Entity]]
+    ) -> list[EntityID]:
         """
-        Get the entity IDs of the specified type
+        Get the entity IDs of the specified types
 
         Parameters
         ----------
-        entity_type : type[Entity]
-            Entity type
+        entity_types : list[type[Entity]]
+            List of entity types
 
         Returns
         -------
@@ -69,7 +71,85 @@ def get_entity_ids_of_type(self, entity_type: type[Entity]) -> list[EntityID]:
         """
         entity_ids: list[EntityID] = []
         for entity in self._world_model.get_entities():
-            if isinstance(entity, entity_type):
+            if any(isinstance(entity, entity_type) for entity_type in entity_types):
                 entity_ids.append(entity.get_id())
 
         return entity_ids
+
+    def get_entities_of_types(self, entity_types: list[type[Entity]]) -> list[Entity]:
+        """
+        Get the entities of the specified types
+
+        Parameters
+        ----------
+        entity_types : list[type[Entity]]
+            List of entity types
+
+        Returns
+        -------
+        list[Entity]
+            Entities
+        """
+        entities: list[Entity] = []
+        for entity in self._world_model.get_entities():
+            if any(isinstance(entity, entity_type) for entity_type in entity_types):
+                entities.append(entity)
+
+        return entities
+
+    def get_distance(self, entity_id1: EntityID, entity_id2: EntityID) -> float:
+        """
+        Get the distance between two entities
+
+        Parameters
+        ----------
+        entity_id1 : EntityID
+            Entity ID 1
+        entity_id2 : EntityID
+            Entity ID 2
+
+        Returns
+        -------
+        float
+            Distance
+
+        Raises
+        ------
+        ValueError
+            If one or both entities are invalid or the location is invalid
+        """
+        entity1: Optional[Entity] = self.get_entity(entity_id1)
+        entity2: Optional[Entity] = self.get_entity(entity_id2)
+        if entity1 is None or entity2 is None:
+            raise ValueError(
+                f"One or both entities are invalid: entity_id1={entity_id1}, entity_id2={entity_id2}, entity1={entity1}, entity2={entity2}"
+            )
+
+        location1_x, location1_y = entity1.get_location()
+        location2_x, location2_y = entity2.get_location()
+        if (
+            location1_x is None
+            or location1_y is None
+            or location2_x is None
+            or location2_y is None
+        ):
+            raise ValueError(
+                f"Invalid location: entity_id1={entity_id1}, entity_id2={entity_id2}, location1_x={location1_x}, location1_y={location1_y}, location2_x={location2_x}, location2_y={location2_y}"
+            )
+
+        distance: float = (
+            (location1_x - location2_x) ** 2 + (location1_y - location2_y) ** 2
+        ) ** 0.5
+
+        return distance
+
+    def get_change_set(self) -> ChangeSet:
+        """
+        Get the change set
+
+        Returns
+        -------
+        ChangeSet
+            Change set
+        """
+        return self._change_set

adf_core_python/core/component/module/algorithm/clustering.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from abc import abstractmethod
+from typing import TYPE_CHECKING
+
+from adf_core_python.core.component.module.abstract_module import AbstractModule
+
+if TYPE_CHECKING:
+    from rcrs_core.entities.entity import Entity
+    from rcrs_core.worldmodel.entityID import EntityID
+
+    from adf_core_python.core.agent.communication.message_manager import MessageManager
+    from adf_core_python.core.agent.develop.develop_data import DevelopData
+    from adf_core_python.core.agent.info.agent_info import AgentInfo
+    from adf_core_python.core.agent.info.scenario_info import ScenarioInfo
+    from adf_core_python.core.agent.info.world_info import WorldInfo
+    from adf_core_python.core.agent.module.module_manager import ModuleManager
+    from adf_core_python.core.agent.precompute.precompute_data import PrecomputeData
+
+
+class Clustering(AbstractModule):
+    def __init__(
+        self,
+        agent_info: AgentInfo,
+        world_info: WorldInfo,
+        scenario_info: ScenarioInfo,
+        module_manager: ModuleManager,
+        develop_data: DevelopData,
+    ) -> None:
+        super().__init__(
+            agent_info, world_info, scenario_info, module_manager, develop_data
+        )
+
+    @abstractmethod
+    def get_cluster_number(self) -> int:
+        pass
+
+    @abstractmethod
+    def get_cluster_index(self, entity_id: EntityID) -> int:
+        pass
+
+    @abstractmethod
+    def get_cluster_entities(self, cluster_index: int) -> list[Entity]:
+        pass
+
+    @abstractmethod
+    def get_cluster_entity_ids(self, cluster_index: int) -> list[EntityID]:
+        pass
+
+    @abstractmethod
+    def calculate(self) -> Clustering:
+        pass
+
+    def precompute(self, precompute_data: PrecomputeData) -> Clustering:
+        super().precompute(precompute_data)
+        return self
+
+    def resume(self, precompute_data: PrecomputeData) -> Clustering:
+        super().resume(precompute_data)
+        return self
+
+    def prepare(self) -> Clustering:
+        super().prepare()
+        return self
+
+    def update_info(self, message_manager: MessageManager) -> Clustering:
+        super().update_info(message_manager)
+        return self

adf_core_python/core/component/module/complex/target_detector.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod
-from typing import TYPE_CHECKING, Generic, TypeVar
+from typing import TYPE_CHECKING, Generic, Optional, TypeVar
 
 from rcrs_core.entities.entity import Entity
 
@@ -35,7 +35,7 @@ def __init__(
         )
 
     @abstractmethod
-    def get_target_entity_id(self) -> EntityID:
+    def get_target_entity_id(self) -> Optional[EntityID]:
         pass
 
     @abstractmethod

adf_core_python/implement/extend_action/default_extend_action_transport.py

@@ -23,6 +23,7 @@
 from adf_core_python.core.component.module.algorithm.path_planning import PathPlanning
 
 
+# TODO: refactor this class
 class DefaultExtendActionTransport(ExtAction):
     def __init__(
         self,
@@ -99,7 +100,7 @@ def calc(self) -> ExtAction:
         agent: AmbulanceTeamEntity = cast(
             AmbulanceTeamEntity, self.agent_info.get_myself()
         )
-        transport_human: Human = self.agent_info.some_one_on_board()
+        transport_human: Optional[Human] = self.agent_info.some_one_on_board()
         if transport_human is not None:
             self.result = self.calc_unload(
                 agent, self._path_planning, transport_human, self._target_entity_id
@@ -134,9 +135,7 @@ def calc_rescue(
 
             target_position = human.get_position()
             if agent_position == target_position:
-                if isinstance(human, Civilian) and (
-                    human.get_buriedness() is not None and human.get_buriedness() > 0
-                ):
+                if isinstance(human, Civilian) and ((human.get_buriedness() or 0) > 0):
                     return ActionLoad(human.get_id())
             else:
                 path = path_planning.get_path(agent_position, target_position)
@@ -176,9 +175,7 @@ def calc_unload(
             if isinstance(position, Refuge):
                 return ActionUnload()
             else:
-                path = path_planning.get_path(
-                    agent_position, self.world_info.get_entity_ids_of_type(Refuge)
-                )
+                path = self.get_nearest_refuge_path(agent, path_planning)
                 if path is not None and len(path) > 0:
                     return ActionMove(path)
 
@@ -191,7 +188,7 @@ def calc_unload(
             human = cast(Human, target_entity)
             if human.get_position() is not None:
                 return self.calc_refuge_action(
-                    agent, path_planning, [human.get_position()], True
+                    agent, path_planning, human.get_position(), True
                 )
             path = self.get_nearest_refuge_path(agent, path_planning)
             if path is not None and len(path) > 0:
@@ -207,7 +204,7 @@ def calc_refuge_action(
         is_unload: bool,
     ) -> Optional[ActionMove | ActionUnload | ActionRest]:
         position = human.get_position()
-        refuges = self.world_info.get_entity_ids_of_type(Refuge)
+        refuges = self.world_info.get_entity_ids_of_types([Refuge])
         size = len(refuges)
 
         if position in refuges:
@@ -242,7 +239,7 @@ def get_nearest_refuge_path(
         self, human: Human, path_planning: PathPlanning
     ) -> list[EntityID]:
         position = human.get_position()
-        refuges = self.world_info.get_entity_ids_of_type(Refuge)
+        refuges = self.world_info.get_entity_ids_of_types([Refuge])
         nearest_path = None
 
         for refuge_id in refuges: