Merge pull request #96 from adf-python/feature/java-module-comm

Feature/java module comm

Author: harrki

.gitignore

@@ -173,3 +173,4 @@ cython_debug/
 # ADF
 agent.log*
 precompute
+!java/lib/src/main/java/adf_core_python/core/agent/precompute

adf_core_python/core/gateway/gateway_agent.py

@@ -2,7 +2,7 @@
 
 from typing import Optional, TYPE_CHECKING, Callable
 
-from rcrs_core.connection import RCRSProto_pb2
+from rcrs_core.connection import RCRSProto_pb2, URN
 
 from adf_core_python.core.agent.info.agent_info import AgentInfo
 from adf_core_python.core.agent.info.scenario_info import ScenarioInfo
@@ -111,3 +111,8 @@ def message_received(self, msg: RCRSProto_pb2) -> None:
 
             self._gateway_modules[c_msg.module_id].set_execute_response(c_msg.result)
             self._gateway_modules[c_msg.module_id].set_is_executed(True)
+
+        if msg.urn == URN.Command.AK_SPEAK:
+            if self.send_msg is None:
+                raise RuntimeError("send_msg is None")
+            self.send_msg(msg)

adf_core_python/launcher.py

@@ -1,4 +1,5 @@
 import argparse
+import resource
 
 from adf_core_python.core.config.config import Config
 from adf_core_python.core.launcher.agent_launcher import AgentLauncher
@@ -11,6 +12,8 @@ def __init__(
         self,
         launcher_config_file: str,
     ) -> None:
+        resource.setrlimit(resource.RLIMIT_NOFILE, (8192, 9223372036854775807))
+
         configure_logger()
 
         self.logger = get_logger(__name__)

java/lib/build.gradle

@@ -39,6 +39,9 @@ dependencies {
     implementation 'org.apache.logging.log4j:log4j-core:2.24.2'
     implementation 'org.apache.logging.log4j:log4j-api:2.24.2'
 
+    //Algorithm
+    implementation 'com.google.common:google-collect:0.5'
+
     testImplementation libs.junit.jupiter
     testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 }

java/lib/src/main/java/adf_core_python/Main.java

@@ -1,6 +1,6 @@
 package adf_core_python;
 
-import adf_core_python.gateway.Gateway;
+import adf_core_python.core.gateway.Gateway;
 
 public class Main {
     public static void main(String[] args) {

java/lib/src/main/java/adf_core_python/agent/develop/DevelopData.java

@@ -1,22 +1,28 @@
-package adf_core_python.agent;
+package adf_core_python.core.agent;
 
-import adf.core.agent.communication.MessageManager;
+import adf.core.agent.communication.standard.bundle.StandardMessageBundle;
+import adf.core.agent.develop.DevelopData;
 import adf.core.agent.info.ScenarioInfo;
 import adf.core.agent.info.WorldInfo;
-import adf_core_python.agent.config.ModuleConfig;
-import adf_core_python.agent.develop.DevelopData;
-import adf_core_python.agent.info.AgentInfo;
-import adf_core_python.agent.module.ModuleManager;
-import adf_core_python.agent.precompute.PrecomputeData;
-import adf_core_python.component.module.AbstractModule;
-import adf_core_python.gateway.mapper.AbstractMapper;
-import adf_core_python.gateway.mapper.MapperDict;
+import adf.core.launcher.ConsoleOutput;
+import adf_core_python.core.agent.communication.MessageManager;
+import adf_core_python.core.agent.communication.standard.StandardCommunicationModule;
+import adf_core_python.core.agent.config.ModuleConfig;
+import adf_core_python.core.agent.info.AgentInfo;
+import adf_core_python.core.agent.module.ModuleManager;
+import adf_core_python.core.agent.precompute.PrecomputeData;
+import adf_core_python.core.component.communication.CommunicationModule;
+import adf_core_python.core.component.module.AbstractModule;
+import adf_core_python.core.gateway.Coordinator;
+import adf_core_python.core.gateway.mapper.AbstractMapper;
+import adf_core_python.core.gateway.mapper.MapperDict;
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import rescuecore2.config.Config;
 import rescuecore2.messages.Command;
+import rescuecore2.messages.Message;
 import rescuecore2.standard.entities.StandardEntityURN;
 import rescuecore2.standard.entities.StandardWorldModel;
 import rescuecore2.worldmodel.ChangeSet;
@@ -25,32 +31,40 @@
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Objects;
 
 public class Agent {
-    private final AgentInfo agentInfo;
-    private final WorldInfo worldInfo;
-    private final ScenarioInfo scenarioInfo;
+    public final AgentInfo agentInfo;
+    public final WorldInfo worldInfo;
+    public final ScenarioInfo scenarioInfo;
     private final ModuleManager moduleManager;
     private final DevelopData developData;
     private final PrecomputeData precomputeData;
     private final MessageManager messageManager;
     private final HashMap<String, AbstractMapper> modules = new HashMap<>();
     private final MapperDict mapperDict;
     private final Logger logger;
+    private final Coordinator coordinator;
+    private CommunicationModule communicationModule;
+    private int ignoreTime;
 
-    public Agent(EntityID entityID, Collection<Entity> entities, ScenarioInfo scenarioInfo, DevelopData developData, ModuleConfig moduleConfig) {
+    public Agent(EntityID entityID, Collection<Entity> entities, ScenarioInfo scenarioInfo, DevelopData developData, ModuleConfig moduleConfig, Coordinator coordinator) {
         StandardWorldModel worldModel = new StandardWorldModel();
         worldModel.addEntities(entities);
         worldModel.index();
 
+        this.ignoreTime = scenarioInfo.getRawConfig()
+                .getIntValue(kernel.KernelConstants.IGNORE_AGENT_COMMANDS_KEY);
+
         this.agentInfo = new AgentInfo(entityID, worldModel);
         this.worldInfo = new WorldInfo(worldModel);
         this.scenarioInfo = scenarioInfo;
         this.developData = developData;
         this.moduleManager = new ModuleManager(this.agentInfo, this.worldInfo, this.scenarioInfo, moduleConfig, this.developData);
+        this.coordinator = coordinator;
 
         String dataStorageName = "";
         StandardEntityURN agentURN = Objects.requireNonNull(this.worldInfo.getEntity(this.agentInfo.getID())).getStandardURN();
@@ -97,13 +111,48 @@ public Class<?> registerModule(@Nonnull String moduleID, @Nonnull String moduleN
     }
 
     public void update(int time, ChangeSet changed, Collection<Command> heard) {
+        worldInfo.setTime(time);
+        worldInfo.merge(changed);
         agentInfo.recordThinkStartTime();
         agentInfo.setTime(time);
+
+        if (time == 1) {
+            if (this.communicationModule != null) {
+                ConsoleOutput.out(ConsoleOutput.State.ERROR,
+                        "[ERROR ] Loader is not found.");
+                ConsoleOutput.out(ConsoleOutput.State.NOTICE,
+                        "CommunicationModule is modified - " + this);
+            } else {
+                this.communicationModule = new StandardCommunicationModule();
+            }
+
+            this.messageManager.registerMessageBundle(new StandardMessageBundle());
+        }
+
+        // agents can subscribe after ignore time
+        if (time >= ignoreTime) {
+            this.messageManager.subscribe(this.agentInfo, this.worldInfo,
+                    this.scenarioInfo);
+
+            if (!this.messageManager.getIsSubscribed()) {
+                int[] channelsToSubscribe = this.messageManager.getChannels();
+                if (channelsToSubscribe != null) {
+                    this.messageManager.setIsSubscribed(true);
+                }
+            }
+        }
+
         agentInfo.setHeard(heard);
         agentInfo.setChanged(changed);
-        worldInfo.setTime(time);
-        worldInfo.merge(changed);
         worldInfo.setChanged(changed);
+
+        this.messageManager.refresh();
+        this.communicationModule.receive(this, this.messageManager);
+
+        this.messageManager.coordinateMessages(this.agentInfo, this.worldInfo,
+                this.scenarioInfo);
+        this.communicationModule.send(this, this.messageManager);
+
         logger.debug("Agent Update (Time: {}, Changed: {}, Heard: {})", agentInfo.getTime(), agentInfo.getChanged(), agentInfo.getHeard());
     }
 
@@ -113,4 +162,12 @@ public Config execModuleMethod(String moduleID, String methodName, Config argume
         logger.debug("Executed Method Result (MethodName: {}, Result: {}", methodName, result);
         return result;
     }
+
+    public EntityID getID() {
+        return this.agentInfo.getID();
+    }
+
+    public void send(Message[] messages) {
+        Arrays.stream(messages).forEach(coordinator::sendMessage);
+    }
 }