Search Module

Overview of the Search Module

The module we will develop here is a Search module that determines information search targets using the KMeansPPClustering module. The clustering module divides the assigned areas among the agents, and a search target is chosen at random from within each assigned area.

Preparing to Implement the Search Module

Note

The following steps assume that the current directory is the project root directory.

First, create the file for the search module.

touch src/<your_team_name>/module/complex/k_means_pp_search.py

Next, implement the search module. Write the following code in k_means_pp_search.py. This will be the template for the search module we implement here.

import random
from typing import Optional, cast

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.component.module.algorithm.clustering import Clustering
from adf_core_python.core.component.module.complex.search import Search
from adf_core_python.core.logger.logger import get_agent_logger
from rcrscore.entities import Building, Entity, Refuge, EntityID


class KMeansPPSearch(Search):
    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
        )
        self._result: Optional[EntityID] = None
        # ロガーの取得
        self._logger = get_agent_logger(
            f"{self.__class__.__module__}.{self.__class__.__qualname__}",
            self._agent_info,
        )

    def calculate(self) -> Search:
        return self

    def get_target_entity_id(self) -> Optional[EntityID]:
        return self._result

Registering the Module

Next, register the module you created. Change the relevant section of config/module.yaml as follows.

DefaultTacticsAmbulanceTeam:
  Search: src.<your_team_name>.module.complex.k_means_pp_search.KMeansPPSearch

DefaultTacticsFireBrigade:
  Search: src.<your_team_name>.module.complex.k_means_pp_search.KMeansPPSearch

DefaultTacticsPoliceForce:
  Search: src.<your_team_name>.module.complex.k_means_pp_search.KMeansPPSearch

Implementing the Module

First, make the KMeansPPClustering module callable.

Add the following code to config/module.yaml.

KMeansPPSearch:
  Clustering: src.<your_team_name>.module.algorithm.k_means_pp_clustering.KMeansPPClustering

Next, make the KMeansPPClustering module callable from the KMeansPPSearch module.

Add the following code to k_means_pp_search.py.

class KMeansPPSearch(Search):
    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
        )
        self._result: Optional[EntityID] = None

        # ロガーの取得
        self._logger = get_agent_logger(
            f"{self.__class__.__module__}.{self.__class__.__qualname__}",
            self._agent_info,
        )

        # クラスタリングモジュールの読み込み
        self._clustering: Clustering = cast(
            Clustering,
            module_manager.get_module(
                # config.yamlに登録したkey
                "KMeansPPSearch.Clustering",
                # 上記のkeyが登録されていなかった場合のデフォルトモジュール
                "adf_core_python.implement.module.algorithm.k_means_clustering.KMeansClustering",
            ),
        )

        # クラスタリングモジュールの登録
        self.register_sub_module(self._clustering)

Then change the calculate method so that it calls the clustering module and determines the search target.

Add the following code to k_means_pp_search.py.

    def calculate(self) -> Search:
        # 自エージェントのエンティティIDを取得
        me: EntityID = self._agent_info.get_entity_id()
        # 自エージェントが所属するクラスターのインデックスを取得
        allocated_cluster_index: int = self._clustering.get_cluster_index(me)
        # クラスター内のエンティティIDを取得
        cluster_entity_ids: list[EntityID] = self._clustering.get_cluster_entity_ids(
            allocated_cluster_index
        )
        # 乱数で選択
        if cluster_entity_ids:
            self._result = random.choice(cluster_entity_ids)

        # ログ出力
        self._logger.info(f"Target entity ID: {self._result}")

        return self

This completes the implementation of the KMeansPPSearch module using the KMeansPPClustering module.

Open two terminals.

Open one terminal and start the simulation server with the following command:

# Terminal A
cd WORKING_DIR/rcrs-server/scripts
./start-comprun.sh -m ../maps/tutorial_ambulance_team_only/map -c ../maps/tutorial_ambulance_team_only/config

Then open another terminal and start the agent:

# Terminal B
cd WORKING_DIR/<your_team_name>
python main.py

Improving the Module

The KMeansPPSearch module uses the clustering module to randomly select search targets from within each assigned area. As a result, it has the following problems.

• The search target changes on every step

  • Because the target changes before the agent reaches it, it is difficult to ever reach the goal

  • Choosing search targets at random from many different places makes efficient search impossible

• Because already searched entities can be selected again as search targets, efficient search is not possible

• Even when there is an unexplored entity nearby, a distant entity may be chosen as the search target

These are some of the problems.

Task

Improve the KMeansPPSearch module and try implementing a module that performs more efficient searching.

Warning

There may be other areas that should be improved in addition to the problems listed here, so you are welcome to improve those as well.

Warning

There are also some improvements to be made in the example program, so please fix them if you have time.

Problem: The search target changes on every step

Hint

Do not change the search target until the once-selected target has been reached

Example Program

    def calculate(self) -> Search:
        # 自エージェントのエンティティIDを取得
        me: EntityID = self._agent_info.get_entity_id()
        # 自エージェントが所属するクラスターのインデックスを取得
        allocated_cluster_index: int = self._clustering.get_cluster_index(me)
        # クラスター内のエンティティIDを取得
        cluster_entity_ids: list[EntityID] = self._clustering.get_cluster_entity_ids(
            allocated_cluster_index
        )

        # 探索対象をすでに選んでいる場合
        if self._result:
            # 自エージェントのいる場所のエンティティIDを取得
            my_position = self._agent_info.get_position_entity_id()
            # 探索対象の場所のエンティティIDを取得
            target_position = self._world_info.get_entity_position(self._result)
            # 自エージェントのいる場所と探索対象の場所が一致している場合、探索対象をリセット
            if my_position == target_position:
                # 探索対象をリセット
                self._result = None

        # 探索対象が未選択の場合
        if not self._result and cluster_entity_ids:
            self._result = random.choice(cluster_entity_ids)

        # ログ出力
        self._logger.info(f"Target entity ID: {self._result}")

        return self

Problem: Already searched entities are selected again as targets

Hint

Record entities that have already been searched in some way so they are not selected again as targets

Example Program

    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
        )
        self._result: Optional[EntityID] = None

        # ロガーの取得
        self._logger = get_agent_logger(
            f"{self.__class__.__module__}.{self.__class__.__qualname__}",
            self._agent_info,
        )

        # クラスタリングモジュールの読み込み
        self._clustering: Clustering = cast(
            Clustering,
            module_manager.get_module(
                # config.yamlに登録したkey
                "KMeansPPSearch.Clustering",
                # 上記のkeyが登録されていなかった場合のデフォルトモジュール
                "adf_core_python.implement.module.algorithm.k_means_clustering.KMeansClustering",
            ),
        )

        # クラスタリングモジュールの要録
        self.register_sub_module(self._clustering)

        # 探索したいエンティティIDのリスト(追加)
        self._search_entity_ids: list[EntityID] = []

    def calculate(self) -> Search:
        # 探索したいエンティティIDのリストが空の場合
        if not self._search_entity_ids:
            # 自エージェントのエンティティIDを取得
            me: EntityID = self._agent_info.get_entity_id()
            # 自エージェントが所属するクラスターのインデックスを取得
            allocated_cluster_index: int = self._clustering.get_cluster_index(me)
            # クラスター内のエンティティIDを取得(変更)
            self._search_entity_ids: list[EntityID] = (
                self._clustering.get_cluster_entity_ids(allocated_cluster_index)
            )

        # 探索対象をすでに選んでいる場合
        if self._result:
            # 自エージェントのいる場所のエンティティIDを取得
            my_position = self._agent_info.get_position_entity_id()
            # 探索対象の場所のエンティティIDを取得
            target_position = self._world_info.get_entity_position(self._result)
            # 自エージェントのいる場所と探索対象の場所が一致している場合、探索対象をリセット
            if my_position == target_position:
                # 探索したいエンティティIDのリストから探索対象を削除
                self._search_entity_ids.remove(self._result)
                # 探索対象をリセット
                self._result = None

        # 探索対象が未選択の場合(変更)
        if not self._result and self._search_entity_ids:
            self._result = random.choice(self._search_entity_ids)

        # ログ出力
        self._logger.info(f"Target entity ID: {self._result}")

        return self

Even when there is an unexplored entity nearby, a distant entity may be chosen as the search target

Hint

Calculate the distance between entities and select the nearest entity as the search target

Example Program

    def calculate(self) -> Search:
        # 探索したいエンティティIDのリストが空の場合
        if not self._search_entity_ids:
            # 自エージェントのエンティティIDを取得
            me: EntityID = self._agent_info.get_entity_id()
            # 自エージェントが所属するクラスターのインデックスを取得
            allocated_cluster_index: int = self._clustering.get_cluster_index(me)
            # クラスター内のエンティティIDを取得
            self._search_entity_ids: list[EntityID] = (
                self._clustering.get_cluster_entity_ids(allocated_cluster_index)
            )

        # 探索対象をすでに選んでいる場合
        if self._result:
            # 自エージェントのいる場所のエンティティIDを取得
            my_position = self._agent_info.get_position_entity_id()
            # 探索対象の場所のエンティティIDを取得
            target_position = self._world_info.get_entity_position(self._result)
            # 自エージェントのいる場所と探索対象の場所が一致している場合、探索対象をリセット
            if my_position == target_position:
                # 探索したいエンティティIDのリストから探索対象を削除
                self._search_entity_ids.remove(self._result)
                # 探索対象をリセット
                self._result = None

        # 探索対象が未選択の場合
        if not self._result and self._search_entity_ids:
            nearest_entity_id: Optional[EntityID] = None
            nearest_distance: float = float("inf")
            me: EntityID = self._agent_info.get_entity_id()
            # 探索対象の中で自エージェントに最も近いエンティティIDを選択(変更)
            for entity_id in self._search_entity_ids:
                distance = self._world_info.get_distance(me, entity_id)
                if distance < nearest_distance:
                    nearest_entity_id = entity_id
                    nearest_distance = distance
            self._result = nearest_entity_id

        # ログ出力
        self._logger.info(f"Target entity ID: {self._result}")

        return self