working on scaling

- dumpModel on site-B
- add coordination and exit logic
- initScene on site-B

ros3rag.common/src/main/java/de/tudresden/inf/st/ros3rag/common/SharedMainParts.java

@@ -26,8 +26,8 @@ public abstract class SharedMainParts<MqttHandler extends SharedMainParts.MqttHa
   private final String TOPIC_REWIND;
   private final String TOPIC_EXIT;
 
-  private static final String TOPIC_SUFFIX_COORDINATOR_STATUS = "status";
-  private static final String TOPIC_SUFFIX_COORDINATOR_COMMAND = "command";
+  public static final String TOPIC_SUFFIX_COORDINATOR_STATUS = "status";
+  public static final String TOPIC_SUFFIX_COORDINATOR_COMMAND = "command";
 
   protected MqttHandler mainHandler;
   protected WorldModel model;
@@ -61,7 +61,7 @@ public abstract class SharedMainParts<MqttHandler extends SharedMainParts.MqttHa
     mainHandler = createMqttHandler().dontSendWelcomeMessage();
     mainHandler.setHost(config.mqttHost);
     if (!mainHandler.waitUntilReady(2, TimeUnit.SECONDS)) {
-      logger.fatal("Could not connect to mqtt handler. Exiting");
+      logger.fatal("Could not connect to " + config.mqttHost + " -> Exiting");
       return;
     }
     CountDownLatch exitCondition = new CountDownLatch(1);
@@ -101,7 +101,7 @@ public abstract class SharedMainParts<MqttHandler extends SharedMainParts.MqttHa
     exitCondition.await();
   }
 
-  private String joinTopics(String... topics) {
+  public static String joinTopics(String... topics) {
     StringBuilder sb = new StringBuilder();
     for (int i = 0; i < topics.length; i++) {
       String topic = topics[i];

ros3rag.common/src/main/resources/config-scaling-mini.yml

 mqttHost: "localhost"
-filenameInitialScene: "src/main/resources/config-scene-a.json"
-coordinatorMqttTopicPrefix: "coordinating/rag-a"
+coordinatorMqttTopicPrefix: "coordinating/rag-"
 topicLogicalUpdate: "place-b/logical-update"
 views: 5
 objects: 7

ros3rag.scaling.a/.gitignore

@@ -3,4 +3,4 @@ src/gen-res/
 src/gen/
 out/
 *.class
-world.svg
+world-a.svg

ros3rag.scaling.a/src/main/java/de/tudresden/inf/st/scaling/a/MainScalingA.java

@@ -2,17 +2,24 @@ package de.tudresden.inf.st.scaling.a;
 
 import de.tudresden.inf.st.jastadd.dumpAst.ast.Dumper;
 import de.tudresden.inf.st.ros3rag.common.ConfigurationScaling;
+import de.tudresden.inf.st.ros3rag.common.SharedMainParts;
 import de.tudresden.inf.st.ros3rag.common.Util;
 import de.tudresden.inf.st.scaling.a.ast.*;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static de.tudresden.inf.st.ros3rag.common.SharedMainParts.TOPIC_SUFFIX_COORDINATOR_STATUS;
+import static de.tudresden.inf.st.ros3rag.common.SharedMainParts.joinTopics;
 
 /**
  * Scaling Case Study - Site A.
@@ -21,27 +28,69 @@ import java.util.Random;
  */
 public class MainScalingA {
   private static final Logger logger = LogManager.getLogger(MainScalingA.class);
+  private static final String TOPIC_EXIT = "place-a/exit";
+
   private CompleteWorld world;
+  private MqttHandler mainHandler;
+  private ConfigurationScaling config;
 
-  public static void main(String[] args) throws IOException {
+  public static void main(String[] args) throws IOException, InterruptedException {
     new MainScalingA().run(args);
   }
 
-  private void run(@SuppressWarnings("unused") String[] args) throws IOException {
+  private void run(@SuppressWarnings("unused") String[] args) throws IOException, InterruptedException {
     logger.info("Hi from A");
     Path pathToConfig = Paths.get("..",
         "ros3rag.common", "src", "main", "resources", "config-scaling-mini.yml");
-    ConfigurationScaling config = Util.parseScalingConfig(pathToConfig.toFile());
+    config = Util.parseScalingConfig(pathToConfig.toFile());
 
     world = createWorld(config);
+    mainHandler = new MqttHandler().dontSendWelcomeMessage().setHost(config.mqttHost);
+    boolean mqttAvailable = mainHandler.waitUntilReady(2, TimeUnit.SECONDS);
+    if (!mqttAvailable) {
+      logger.fatal("Could not connect to " + config.mqttHost + " -> Exiting.");
+      return;
+    }
+
     generateAndSetViews();
-    Dumper.read(world).dumpAsSVG(Paths.get("world.svg"));
+    Dumper.read(world).dumpAsSVG(Paths.get("world-a.svg"));
+
+    logger.info("Using coordinator logic");
+    final String coordinatorPrefix = config.coordinatorMqttTopicPrefix + "a";
+    mainHandler.newConnection(SharedMainParts.joinTopics(coordinatorPrefix,
+            SharedMainParts.TOPIC_SUFFIX_COORDINATOR_COMMAND),
+        bytes -> {
+          connectMyScenes();
+          mainHandler.publish(joinTopics(coordinatorPrefix, TOPIC_SUFFIX_COORDINATOR_STATUS),
+              "ready".getBytes(StandardCharsets.UTF_8));
+        });
+
+    Runtime.getRuntime().addShutdownHook(new Thread(this::close));
+
+    CountDownLatch exitCondition = new CountDownLatch(1);
+    mainHandler.newConnection(TOPIC_EXIT, bytes -> {
+      logger.info("Got exit command");
+      exitCondition.countDown();
+      logger.debug("exit latch count = {}",
+          exitCondition.getCount());
+    });
 
+    mainHandler.publish(joinTopics(coordinatorPrefix, TOPIC_SUFFIX_COORDINATOR_STATUS),
+        "up".getBytes(StandardCharsets.UTF_8));
+
+    exitCondition.await();
+  }
+
+  private void connectMyScenes() {
     for (int viewIndex = 0; viewIndex < world.getConfig().getNumberOfViews(); viewIndex++) {
+      try {
         world.connectView(
-          Util.mqttUri(config.topicLogicalUpdate, config.mqttHost),
+            Util.mqttUri(joinTopics(config.topicLogicalUpdate, String.valueOf(viewIndex)), config.mqttHost),
             viewIndex,
             true);
+      } catch (IOException e) {
+        e.printStackTrace();
+      }
     }
   }
 
@@ -93,4 +142,10 @@ public class MainScalingA {
       world.addView(view);
     }
   }
+
+  private void close() {
+    logger.info("Exiting ...");
+    mainHandler.close();
+    world.ragconnectCloseConnections();
+  }
 }

ros3rag.scaling.b/.gitignore

@@ -3,3 +3,4 @@ src/gen-res/
 src/gen/
 out/
 *.class
+world-b.svg

ros3rag.scaling.b/src/main/java/de/tudresden/inf/st/scaling/b/MainScalingB.java

 package de.tudresden.inf.st.scaling.b;
 
+import de.tudresden.inf.st.jastadd.dumpAst.ast.Dumper;
 import de.tudresden.inf.st.ros3rag.common.ConfigurationScaling;
+import de.tudresden.inf.st.ros3rag.common.SharedMainParts;
 import de.tudresden.inf.st.ros3rag.common.Util;
 import de.tudresden.inf.st.scaling.b.ast.*;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static de.tudresden.inf.st.ros3rag.common.SharedMainParts.TOPIC_SUFFIX_COORDINATOR_STATUS;
+import static de.tudresden.inf.st.ros3rag.common.SharedMainParts.joinTopics;
+
 /**
  * Scaling Case Study - Site B.
  *
@@ -19,37 +27,137 @@ import java.util.stream.IntStream;
  */
 public class MainScalingB {
   private static final Logger logger = LogManager.getLogger(MainScalingB.class);
+  private static final String TOPIC_EXIT = "place-b/exit";
+  private static final String TOPIC_MODEL = "place-b/model";
+  private static final String TOPIC_MODEL_SVG_PATH = "place-b/model/svg/path";
+
   private WorldModelB world;
+  private MqttHandler mainHandler;
+  private ConfigurationScaling config;
 
-  public static void main(String[] args) throws IOException {
+  public static void main(String[] args) throws IOException, InterruptedException {
     new MainScalingB().run(args);
   }
 
-  private void run(@SuppressWarnings("unused") String[] args) throws IOException {
+  private void run(@SuppressWarnings("unused") String[] args) throws IOException, InterruptedException {
     logger.info("Hi from B");
     Path pathToConfig = Paths.get("..",
         "ros3rag.common", "src", "main", "resources", "config-scaling-mini.yml");
-    ConfigurationScaling config = Util.parseScalingConfig(pathToConfig.toFile());
+    config = Util.parseScalingConfig(pathToConfig.toFile());
 
     world = new WorldModelB();
-    generateWorld(config);
+    mainHandler = new MqttHandler().dontSendWelcomeMessage().setHost(config.mqttHost);
+    boolean mqttAvailable = mainHandler.waitUntilReady(2, TimeUnit.SECONDS);
+    if (!mqttAvailable) {
+      logger.fatal("Could not connect to " + config.mqttHost + " -> Exiting.");
+      return;
+    }
+
+    generateWorld();
+    dumpModel();
+
+    logger.info("Using coordinator logic");
+    final String coordinatorPrefix = config.coordinatorMqttTopicPrefix + "b";
+    mainHandler.newConnection(joinTopics(coordinatorPrefix,
+            SharedMainParts.TOPIC_SUFFIX_COORDINATOR_COMMAND),
+        bytes -> {
+          initMyScene();
+          mainHandler.publish(joinTopics(coordinatorPrefix, TOPIC_SUFFIX_COORDINATOR_STATUS),
+              "ready".getBytes(StandardCharsets.UTF_8));
+        });
+
+    Runtime.getRuntime().addShutdownHook(new Thread(this::close));
+
+    CountDownLatch exitCondition = new CountDownLatch(1);
+    mainHandler.newConnection(TOPIC_EXIT, bytes -> {
+      logger.info("Got exit command");
+      exitCondition.countDown();
+      logger.debug("exit latch count = {}",
+          exitCondition.getCount());
+    });
+    mainHandler.newConnection(TOPIC_MODEL, bytes -> dumpModel());
+
+    mainHandler.publish(joinTopics(coordinatorPrefix, TOPIC_SUFFIX_COORDINATOR_STATUS),
+        "up".getBytes(StandardCharsets.UTF_8));
+
+    exitCondition.await();
+  }
+
+  private void initMyScene() {
+    Scene myNewScene = new Scene();
+
+    int pcc = 0; // position-coordination-counter
+    // locations
+    for (Region region : world.getRegionList()) {
+      if (region.getName().startsWith("Collab")) {
+        // exactly one collaboration zone
+        CollaborationZone cz = new CollaborationZone()
+            .setName("P-" + region.getName())
+            .setPosition(Position.of(pcc, pcc, pcc))
+            .setOrientation(Orientation.of(0, 0, 0, 1))
+            .setSize(Size.of(0.1, 0.1, 0));
+        myNewScene.addDropOffLocation(cz);
+        pcc += 1;
+      } else {
+        for (String locationName : region.locationNamesAsList()) {
+          DropOffLocation location = new DropOffLocation()
+              .setName(locationName)
+              .setPosition(Position.of(pcc, pcc, pcc))
+              .setOrientation(Orientation.of(0, 0, 0, 1))
+              .setSize(Size.of(0.1, 0.1, 0));
+          myNewScene.addDropOffLocation(location);
+          pcc += 1;
+        }
+      }
+    }
+
+    // all objects at start
+    for (int objectIndex = 0; objectIndex < config.objects; objectIndex++) {
+      DropOffLocation location = myNewScene.resolveObjectOfInterest("P-Start" + objectIndex).asDropOffLocation();
+      MovableObject obj = new MovableObject()
+          .setName("O" + objectIndex)
+          .setPosition(Position.of(location.getPosition().getX(),
+              location.getPosition().getY(),
+              location.getPosition().getZ()))
+          .setOrientation(Orientation.of(0, 0, 0, 1))
+          .setSize(Size.of(0, 0, 0));
+      myNewScene.addMovableObject(obj);
     }
 
-  private void generateWorld(ConfigurationScaling config) throws IOException {
+    // robot objects
+    for (int robotIndex = 0; robotIndex < config.robots; robotIndex++) {
+      myNewScene.addRobotObject(new RobotObject()
+          .setName("ARM" + robotIndex)
+          .setActive(false)
+          .setPosition(new Position())
+          .setOrientation(new Orientation())
+          .setSize(new Size()));
+    }
+
+    world.setMyScene(myNewScene);
+  }
+
+  private void generateWorld() throws IOException {
     // start region
-    world.addRegion(new Region().setName("Start").setLocationNames(locationsFor("Start", config)));
+    world.addRegion(new Region().setName("Start").setLocationNames(locationsFor("Start")));
 
     // robots
     for (int robotIndex = 0; robotIndex < config.robots; robotIndex++) {
       // region of the robot
       String regionOfRobot = "Region" + robotIndex;
-      world.addRegion(new Region().setName(regionOfRobot).setLocationNames(locationsFor(regionOfRobot, config)));
+      world.addRegion(new Region().setName(regionOfRobot).setLocationNames(locationsFor(regionOfRobot)));
+
+      // region of collaboration zones
+      if (robotIndex != config.robots - 1) {
+        String collaborationRegion = "Collab" + robotIndex;
+        world.addRegion(new Region().setName(collaborationRegion).setLocationNames("P-" + collaborationRegion));
+      }
 
       Robot robot = new Robot().setName("ARM" + robotIndex);
 
       // reachability
       CanReachObjectOfInterestWrapper wrapper = new CanReachObjectOfInterestWrapper();
-      // each robot arm can reach the region with the same index as the arm (and the start region, for the first robot)
+      // each robot arm can reach the location of its region (and the start region, for the first robot)
       if (robotIndex == 0) {
         for (String location : world.findRegion("Start").locationNamesAsList()) {
           wrapper.addCanReachObjectOfInterest(new CanReachObjectOfInterest(location));
@@ -58,6 +166,15 @@ public class MainScalingB {
       for (String location : world.findRegion(regionOfRobot).locationNamesAsList()) {
         wrapper.addCanReachObjectOfInterest(new CanReachObjectOfInterest(location));
       }
+      // add reachability for collaboration zones
+      if (robotIndex != 0) {
+        // reachability to previous collaboration zone
+        wrapper.addCanReachObjectOfInterest(new CanReachObjectOfInterest("P-Collab" + (robotIndex - 1)));
+      }
+      if (robotIndex != config.robots - 1) {
+        // reachability to its collaboration zone
+        wrapper.addCanReachObjectOfInterest(new CanReachObjectOfInterest("P-Collab" + (robotIndex)));
+      }
       robot.setCanReachObjectOfInterestWrapper(wrapper);
       world.addRobot(robot);
     }
@@ -65,16 +182,38 @@ public class MainScalingB {
     // other scenes
     for (int sceneIndex = 0; sceneIndex < config.views; sceneIndex++) {
       world.addOtherScene(new LogicalScene());
-      world.connectOtherScene(config.topicLogicalUpdate, sceneIndex);
     }
+    world.connectOtherScene(Util.mqttUri(joinTopics(config.topicLogicalUpdate, "#"), config.mqttHost));
 
     // my scene
     world.setMyScene(new Scene());
   }
 
-  private String locationsFor(String regionName, ConfigurationScaling config) {
+  private String locationsFor(String regionName) {
     return IntStream.range(0, config.objects)
         .mapToObj(index -> "P-" + regionName + index)
         .collect(Collectors.joining(","));
   }
+
+  private void dumpModel() {
+    try {
+      String filename = world.dumpAst(builder -> {
+        builder.excludeChildren("Orientation", "Size");
+        builder.excludeRelations("ContainedInRegion");
+        builder.includeNonterminalAttributes("LogicalScene", "diffScenes", "diffToOperations");
+        builder.includeAttributes("realRegion", "computeOperations");
+        builder.includeNullNodes();
+      });
+      mainHandler.publish(TOPIC_MODEL_SVG_PATH, filename.getBytes(StandardCharsets.UTF_8));
+    } catch (Exception e) {
+      logger.catching(e);
+    }
+  }
+
+  private void close() {
+    logger.info("Exiting ...");
+    mainHandler.close();
+    world.ragconnectCloseConnections();
+  }
+
 }