司令塔の実装

adf_core_python/core/agent/module/module_manager.py

@@ -4,6 +4,8 @@
 from typing import TYPE_CHECKING, Any, Optional
 
 from adf_core_python.core.component.action.extend_action import ExtendAction
+from adf_core_python.core.component.centralized.command_executor import CommandExecutor
+from adf_core_python.core.component.centralized.command_picker import CommandPicker
 from adf_core_python.core.component.communication.channel_subscriber import (
     ChannelSubscriber,
 )
@@ -189,6 +191,58 @@ def get_message_coordinator(
             f"Message coordinator {class_name} is not a subclass of MessageCoordinator"
         )
 
+    def get_command_executor(
+        self, command_executor_name: str, default_command_executor_name: str
+    ) -> CommandExecutor:
+        class_name = self._module_config.get_value_or_default(
+            command_executor_name, default_command_executor_name
+        )
+
+        command_executor_class: type = self._load_module(class_name)
+
+        instance = self._executors.get(command_executor_name)
+        if instance is not None:
+            return instance
+
+        if issubclass(command_executor_class, CommandExecutor):
+            instance = command_executor_class(
+                self._agent_info,
+                self._world_info,
+                self._scenario_info,
+                self,
+                self._develop_data,
+            )
+            self._executors[command_executor_name] = instance
+            return instance
+
+        raise RuntimeError(f"Command executor {class_name} is not a subclass of object")
+
+    def get_command_picker(
+        self, command_picker_name: str, default_command_picker_name: str
+    ) -> CommandPicker:
+        class_name = self._module_config.get_value_or_default(
+            command_picker_name, default_command_picker_name
+        )
+
+        command_picker_class: type = self._load_module(class_name)
+
+        instance = self._pickers.get(command_picker_name)
+        if instance is not None:
+            return instance
+
+        if issubclass(command_picker_class, CommandPicker):
+            instance = command_picker_class(
+                self._agent_info,
+                self._world_info,
+                self._scenario_info,
+                self,
+                self._develop_data,
+            )
+            self._pickers[command_picker_name] = instance
+            return instance
+
+        raise RuntimeError(f"Command picker {class_name} is not a subclass of object")
+
     def _load_module(self, class_name: str) -> type:
         module_name, module_class_name = class_name.rsplit(".", 1)
         module = importlib.import_module(module_name)

adf_core_python/core/component/centralized/command_executor.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Generic, Optional, TypeVar
+
+if TYPE_CHECKING:
+    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
+
+from adf_core_python.core.agent.action.action import Action
+from adf_core_python.core.component.communication.communication_message import (
+    CommunicationMessage,
+)
+
+T = TypeVar("T", bound=CommunicationMessage)
+
+
+class CommandExecutor(ABC, Generic[T]):
+    def __init__(
+        self,
+        agent_info: AgentInfo,
+        world_info: WorldInfo,
+        scenario_info: ScenarioInfo,
+        module_manager: ModuleManager,
+        develop_data: DevelopData,
+    ) -> None:
+        self._agent_info = agent_info
+        self._world_info = world_info
+        self._scenario_info = scenario_info
+        self._module_manager = module_manager
+        self._develop_data = develop_data
+
+        self._result: Optional[Action] = None
+
+        self._count_precompute: int = 0
+        self._count_prepare: int = 0
+        self._count_resume: int = 0
+        self._count_update_info: int = 0
+        self._count_update_info_current_time: int = 0
+
+    @abstractmethod
+    def set_command(self, command: T) -> CommandExecutor:
+        pass
+
+    @abstractmethod
+    def calculate(self) -> CommandExecutor:
+        pass
+
+    def get_action(self) -> Optional[Action]:
+        return self._result
+
+    def precompute(self, precompute_data: PrecomputeData) -> CommandExecutor:
+        self._count_precompute += 1
+        return self
+
+    def prepare(self) -> CommandExecutor:
+        self._count_prepare += 1
+        return self
+
+    def resume(self, precompute_data: PrecomputeData) -> CommandExecutor:
+        self._count_resume += 1
+        return self
+
+    def update_info(self, message_manager: MessageManager) -> CommandExecutor:
+        if self._count_update_info_current_time != self._agent_info.get_time():
+            self._count_update_info_current_time = self._agent_info.get_time()
+            self._count_update_info += 1
+        return self
+
+    def get_count_precompute(self) -> int:
+        return self._count_precompute
+
+    def get_count_prepare(self) -> int:
+        return self._count_prepare
+
+    def get_count_resume(self) -> int:
+        return self._count_resume
+
+    def get_count_update_info(self) -> int:
+        return self._count_update_info
+
+    def reset_count_precompute(self) -> None:
+        self._count_precompute = 0
+
+    def reset_count_prepare(self) -> None:
+        self._count_prepare = 0
+
+    def reset_count_resume(self) -> None:
+        self._count_resume = 0
+
+    def reset_count_update_info(self) -> None:
+        self._count_update_info = 0

adf_core_python/core/component/centralized/command_picker.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from rcrs_core.worldmodel.entityID import EntityID
+
+if TYPE_CHECKING:
+    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.communication.message_manager import MessageManager
+from adf_core_python.core.agent.precompute.precompute_data import PrecomputeData
+from adf_core_python.core.component.communication.communication_message import (
+    CommunicationMessage,
+)
+
+
+class CommandPicker(ABC):
+    def __init__(
+        self,
+        agent_info: AgentInfo,
+        world_info: WorldInfo,
+        scenario_info: ScenarioInfo,
+        module_manager: ModuleManager,
+        develop_data: DevelopData,
+    ) -> None:
+        self._agent_info = agent_info
+        self._world_info = world_info
+        self._scenario_info = scenario_info
+        self._module_manager = module_manager
+        self._develop_data = develop_data
+        self._count_precompute: int = 0
+        self._count_prepare: int = 0
+        self._count_resume: int = 0
+        self._count_update_info: int = 0
+        self._count_update_info_current_time: int = 0
+
+    @abstractmethod
+    def set_allocator_result(
+        self, allocation_data: dict[EntityID, EntityID]
+    ) -> CommandPicker:
+        pass
+
+    @abstractmethod
+    def calculate(self) -> CommandPicker:
+        pass
+
+    @abstractmethod
+    def get_result(self) -> list[CommunicationMessage]:
+        pass
+
+    def precompute(self, precompute_data: PrecomputeData) -> CommandPicker:
+        self._count_precompute += 1
+        return self
+
+    def prepare(self) -> CommandPicker:
+        self._count_prepare += 1
+        return self
+
+    def resume(self, precompute_data: PrecomputeData) -> CommandPicker:
+        self._count_resume += 1
+        return self
+
+    def update_info(self, message_manager: MessageManager) -> CommandPicker:
+        if self._count_update_info_current_time != self._agent_info.get_time():
+            self._count_update_info_current_time = self._agent_info.get_time()
+            self._count_update_info = 0
+        self._count_update_info += 1
+        return self
+
+    def get_count_precompute(self) -> int:
+        return self._count_precompute
+
+    def get_count_prepare(self) -> int:
+        return self._count_prepare
+
+    def get_count_resume(self) -> int:
+        return self._count_resume
+
+    def get_count_update_info(self) -> int:
+        return self._count_update_info
+
+    def get_count_update_info_current_time(self) -> int:
+        return self._count_update_info_current_time
+
+    def reset_count_precompute(self) -> None:
+        self._count_precompute = 0
+
+    def reset_count_prepare(self) -> None:
+        self._count_prepare = 0
+
+    def reset_count_resume(self) -> None:
+        self._count_resume = 0
+
+    def reset_count_update_info(self) -> None:
+        self._count_update_info = 0

adf_core_python/core/component/tactics/tactics_center.py

@@ -3,6 +3,8 @@
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Any, Optional
 
+from adf_core_python.core.component.centralized.command_picker import CommandPicker
+
 if TYPE_CHECKING:
     from adf_core_python.core.agent.communication.message_manager import MessageManager
     from adf_core_python.core.agent.develop.develop_data import DevelopData
@@ -18,7 +20,7 @@ class TacticsCenter(ABC):
     def __init__(self, parent: Optional[TacticsCenter] = None) -> None:
         self._parent = parent
         self._modules: list[AbstractModule] = []
-        self._command_pickers: list[Any] = []
+        self._command_pickers: list[CommandPicker] = []
 
     @abstractmethod
     def initialize(