1

pom.xml

@@ -42,6 +42,11 @@
         <artifactId>Java-WebSocket</artifactId>
         <version>1.5.4</version>
       </dependency>
+      <dependency>
+        <groupId>org.eclipse.paho</groupId>
+        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+        <version>1.2.5</version>
+      </dependency>
     </dependencies>
 
     <build>

src/main/java/com/iflytop/digester/DigestionTaskTheadManager.java

@@ -1,16 +1,40 @@
 package com.iflytop.digester;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iflytop.digester.deviceinstance.HeatingTurntableSlotTube;
 import com.iflytop.digester.model.MdbDigestionTask;
 import jakarta.annotation.PostConstruct;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
-
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+
 @Component
 public class DigestionTaskTheadManager {
     // Singleton instance
     private static DigestionTaskTheadManager instance = null;
     // Task list
     private List<DigestionTaskThread> tasks;
+    // Mqtt client
+    private MqttClient client;
+
+    @Value("${mqtt-broker.uri}")
+    private String mqttBrokerUri;
+    @Value("${mqtt-broker.username}")
+    private String mqttBrokerUsername;
+    @Value("${mqtt-broker.password}")
+    private String mqttBrokerPassword;
+    @Value("${mqtt-broker.clientId}")
+    private String mqttBrokerClientId;
+    @Value("${mqtt-broker.my-topic}")
+    private String mqttBrokerMyTopic;
+    @Value("${mqtt-broker.transbot-topic}")
+    private String mqttBrokerTransbotTopic;
 
     // Get instance
     public static DigestionTaskTheadManager getInstance() {
@@ -21,12 +45,101 @@ public class DigestionTaskTheadManager {
     public void init() {
         this.tasks = new ArrayList<>();
         instance = this;
+        this.setupMqttBroker();
+    }
+
+    // setup mqtt broker
+    private void setupMqttBroker() {
+        this.client = null;
+        try {
+            this.client = new MqttClient(this.mqttBrokerUri, this.mqttBrokerClientId, new MemoryPersistence());
+        } catch (MqttException e) {
+            throw new RuntimeException(e);
+        }
+        this.client.setCallback(new MqttCallback() {
+            public void connectionLost(Throwable cause) {
+                System.out.println("connectionLost: " + cause.getMessage());
+            }
+            public void messageArrived(String topic, MqttMessage message) {
+                DigestionTaskTheadManager.this.handleOnMessageArrived(topic, message);
+            }
+            public void deliveryComplete(IMqttDeliveryToken token) {
+                System.out.println("deliveryComplete---------" + token.isComplete());
+            }
+        });
+        try {
+            MqttConnectOptions options = new MqttConnectOptions();
+            options.setUserName(this.mqttBrokerUsername);
+            options.setPassword(this.mqttBrokerPassword.toCharArray());
+            client.connect(options);
+        } catch (MqttException e) {
+            throw new RuntimeException(e);
+        }
+
+        try {
+            client.subscribe(this.mqttBrokerMyTopic, 2);
+        } catch (MqttException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    // Handle on message arrived
+    private void handleOnMessageArrived(String topic, MqttMessage message) {
+        String content = new String(message.getPayload());
+        JsonNode actionJsonTree = null;
+        try {
+            ObjectMapper jsonMapper = new ObjectMapper();
+            actionJsonTree = jsonMapper.readTree(content);
+        } catch (JsonProcessingException e) {
+            // ignore invalid message
+        }
+
+        assert actionJsonTree != null;
+        String taskId = actionJsonTree.get("taskId").asText();
+        String actionName = actionJsonTree.get("action").asText();
+        Map<String, Object> actionParams = new HashMap<>();
+        if ( actionJsonTree.has("params") ) {
+            var paramsJsonTree = actionJsonTree.get("params");
+            for ( JsonNode paramNode : paramsJsonTree ) {
+                actionParams.put(paramNode.get("key").asText(), paramNode.get("value").asText());
+            }
+        }
+
+        DigestionTaskThread task = null;
+        for ( var t : this.tasks ) {
+            if ( t.getTaskId().equals(taskId) ) {
+                task = t;
+                break;
+            }
+        }
+        if ( null == task ) {
+            // ignore invalid task
+            return ;
+        }
+        task.executeAction(actionName, actionParams);
+    }
+
+    // Send message to trans bot
+    public void sendMessageToTransBot(String message) {
+        try {
+            MqttMessage mqttMessage = new MqttMessage(message.getBytes());
+            mqttMessage.setQos(2);
+            this.client.publish(this.mqttBrokerTransbotTopic, mqttMessage);
+        } catch (MqttException e) {
+            throw new RuntimeException(e);
+        }
     }
 
     // Start task
     public void startTask( MdbDigestionTask taskModel ) {
         var task = new DigestionTaskThread(taskModel);
-        tasks.add(task);
+        task.setFinishCallback(this::handleOnTaskFinished);
+        this.tasks.add(task);
         task.start();
     }
+
+    // Handle task finished
+    private void handleOnTaskFinished( DigestionTaskThread task ) {
+        this.tasks.remove(task);
+    }
 }

src/main/java/com/iflytop/digester/DigestionTaskThread.java

@@ -8,7 +8,14 @@ import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+
 public class DigestionTaskThread extends Thread {
+    // 任务完成回调
+    public interface FinishCallback {
+        void callback( DigestionTaskThread task );
+    }
+
     // logger
     public static final Logger LOG = LoggerFactory.getLogger(DigestionTaskThread.class);
     // 消解任务Model
@@ -19,6 +26,14 @@ public class DigestionTaskThread extends Thread {
     private HeatingTurntableSlot heatingSlot;
     // 异常试管索引列表
     private List<Integer> errorTubeIndexes;
+    // 任务完成回调
+    private FinishCallback finishCallback;
+    // 试管架放入等待锁
+    private final Object tubeRackPutInWaitLock = new Object();
+    // 试管架取出等待锁
+    private final Object tubeRackTakeOutWaitLock = new Object();
+    // 异常处理线程
+    private Thread errorProcessThread;
 
     // 消解任务
     public DigestionTaskThread(MdbDigestionTask taskModel) {
@@ -26,6 +41,29 @@ public class DigestionTaskThread extends Thread {
         this.solution = UfActiveRecord.findOne(MdbDigestionSolution.class, taskModel.digestionId);
     }
 
+    // 设置任务完成回调
+    public void setFinishCallback(FinishCallback finishCallback) {
+        this.finishCallback = finishCallback;
+    }
+
+    // 获取任务ID
+    public String getTaskId() {
+        return this.taskModel.taskId;
+    }
+
+    // 执行动作
+    public void executeAction(String name, Map<String,Object> params) {
+        if ( "TubeRackPutInDone".equals(name) ) {
+            synchronized ( this.tubeRackPutInWaitLock ) {
+                this.tubeRackPutInWaitLock.notifyAll();
+            }
+        } else if ( "TubeRackTakeOutDone".equals(name) ) {
+            synchronized ( this.tubeRackTakeOutWaitLock ) {
+                this.tubeRackTakeOutWaitLock.notifyAll();
+            }
+        }
+    }
+
     @Override
     public void run() {
         try {
@@ -44,30 +82,27 @@ public class DigestionTaskThread extends Thread {
             this.executeRound(lastRound);
             // 检查试管
             this.tubeCheck();
-
-            // 异常处理执行
-            var errorRound = this.solution.getErrorRounds();
-            for ( int i=0; i<errorRound.maxRounds; i++ ) {
-                this.executeErrorRound(errorRound);
-                if (this.errorTubeIndexes.isEmpty()) {
-                    break;
-                }
-            }
-
+            // 启动异常处理线程
+            this.startErrorProcessThread();
             // 消解结束
             this.finish();
         } catch (Exception e) {
-            this.updateTaskStatus("error", e.getMessage());
+            this.updateTaskStatus("Error", e.getMessage());
         }
     }
 
     // 消解准备
     private void prepare() {
-        this.updateTaskStatus("prepare", "消解任务准备中");
+        this.updateTaskStatus("Prepare", "消解任务准备中");
 
         var device = Device.getInstance();
+
         // 分配加热位
         this.heatingSlot = device.heatingTurntable.allocSlot(this.taskModel.tubeRackNo);
+        this.heatingSlot.setTubes(this.taskModel.getTubes());
+        this.taskModel.heatingSlotIndex = this.heatingSlot.index;
+        this.taskModel.save();
+
         // 打开门
         device.door.open();
         // 等待放入试管架
@@ -79,20 +114,29 @@ public class DigestionTaskThread extends Thread {
     // 消解结束
     private void finish() {
         var device = Device.getInstance();
+
+        // 余温加热
+        this.heatingSlot.heating(100, 10);
+        // 移动到加液盘
+        device.transferArm.moveTubeRackToLiquidPlate(this.heatingSlot.index);
         // 打开门
         device.door.open();
-        // 等待取出试管架
+        // 等待取出
         this.waitForTubeRackTakeOut();
         // 关闭门
         device.door.close();
         // 释放加热位
         this.heatingSlot.setTubeRackNo(null);
 
-        this.updateTaskStatus("finish", "消解任务结束");
+        // 更新任务状态
+        this.updateTaskStatus("Finish", "消解任务结束");
+
+        // 任务完成回调
+        this.finishCallback.callback(this);
     }
 
     // 检查试管
-    private void tubeCheck() {
+    private void tubeCheck() throws InterruptedException {
         var device = Device.getInstance();
         // 移动到加液盘
         device.transferArm.moveTubeRackToLiquidPlate(this.heatingSlot.index);
@@ -100,6 +144,27 @@ public class DigestionTaskThread extends Thread {
         this.errorTubeIndexes = this.takeShotAndCheckErrorTubes();
         // 将异常试管放入异常处理区域
         device.transferArm.moveTubesToErrorSlot(this.errorTubeIndexes);
+        // 将正常试管放入加热转盘
+        device.transferArm.moveTubeRackToHeatingTurntable(this.heatingSlot.index);
+    }
+
+    // 启动异常处理线程
+    private void startErrorProcessThread() {
+        this.errorProcessThread = new Thread(() -> {
+            try {
+                // 异常处理执行
+                var errorRound = this.solution.getErrorRounds();
+                for ( int i=0; i<errorRound.maxRounds; i++ ) {
+                    this.executeErrorRound(errorRound);
+                    if (this.errorTubeIndexes.isEmpty()) {
+                        break;
+                    }
+                }
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        this.errorProcessThread.start();
     }
 
     // 执行消解每轮配置
@@ -110,6 +175,7 @@ public class DigestionTaskThread extends Thread {
         var tubes = this.heatingSlot.getExistTubeIndexes();
         var liquidAddition = device.liquidAddition;
         for ( var liquid : digestionRound.liquids ) {
+            this.updateTaskStatus("Liquid", String.format("加液 %s %d", liquid.type, liquid.volume));
             liquidAddition.addLiquidToTubes(tubes, liquid.type, liquid.volume);
         }
 
@@ -117,10 +183,9 @@ public class DigestionTaskThread extends Thread {
         device.transferArm.moveTubeRackToHeatingTurntable(this.heatingSlot.index);
 
         // 加热
-//        var heatingTurntable = HeatingTurntableInstance.getInstance();
         for ( var heating : digestionRound.heatings ) {
+            this.updateTaskStatus("heating", String.format("加热 %d℃ %d分钟", heating.temperature, heating.duration));
             this.heatingSlot.heating(heating.temperature, heating.duration);
-//            heatingTurntable.heating(1, heating.temperature, heating.duration);
         }
 
         // 移至加液区， 以便执行下一轮
@@ -137,6 +202,7 @@ public class DigestionTaskThread extends Thread {
         // 加液
         var liquidAddition = device.liquidAddition;
         for ( var liquid : errorRound.liquids ) {
+            this.updateTaskStatus("liquid", String.format("异常加液 %s %d", liquid.type, liquid.volume));
             liquidAddition.addLiquidToTubes(this.errorTubeIndexes, liquid.type, liquid.volume);
         }
 
@@ -144,10 +210,9 @@ public class DigestionTaskThread extends Thread {
         device.transferArm.moveTubeRackToHeatingTurntable(this.heatingSlot.index);
 
         // 加热
-//        var heatingTurntable = HeatingTurntableInstance.getInstance();
         for ( var heating : errorRound.heatings ) {
+            this.updateTaskStatus("heating", String.format("异常加热 %d℃ %d分钟", heating.temperature, heating.duration));
             this.heatingSlot.heating(heating.temperature, heating.duration);
-//            heatingTurntable.heating(1, heating.temperature, heating.duration);
         }
 
         // 移至加液区
@@ -164,12 +229,28 @@ public class DigestionTaskThread extends Thread {
 
     // 等待放入试管架
     private void waitForTubeRackPutIn() {
-        // 等待放入试管架
+        this.updateTaskStatus("TubeRackPutInWait", "等待放入试管架");
+        synchronized ( this.tubeRackPutInWaitLock ) {
+            try {
+                this.tubeRackPutInWaitLock.wait();
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        this.updateTaskStatus("TubeRackPutInWaitDone", "试管架已放入");
     }
 
     // 等待取出试管架
     private void waitForTubeRackTakeOut() {
-        // 等待取出试管架
+        this.updateTaskStatus("TubeRackTakeOutWait", "等待取出试管架");
+        synchronized ( this.tubeRackTakeOutWaitLock ) {
+            try {
+                this.tubeRackTakeOutWaitLock.wait();
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        this.updateTaskStatus("TubeRackTakeOutWaitDone", "试管架已取出");
     }
 
     // 拍照检查试管
@@ -184,9 +265,9 @@ public class DigestionTaskThread extends Thread {
 
     // 更新任务状态
     private void updateTaskStatus(String status, String message) {
-        LOG.info("[Digestion Task : {}] ({}) {}", taskModel.taskId, status, message);
-        taskModel.status = status;
-        taskModel.message = message;
-        taskModel.save();
+        LOG.info("[Digestion Task : {}] ({}) {}", taskModel.id, status, message);
+        this.taskModel.status = status;
+        this.taskModel.message = message;
+        this.taskModel.save();
     }
 }

src/main/java/com/iflytop/digester/deviceinstance/HeatingTurntableSlot.java

@@ -1,6 +1,10 @@
 package com.iflytop.digester.deviceinstance;
+import com.iflytop.digester.underframework.UfCmdSnippetExecutor;
+
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+
 public class HeatingTurntableSlot {
     // 槽位索引
     public Integer index = -1;
@@ -13,27 +17,53 @@ public class HeatingTurntableSlot {
     // 加热时长
     public Integer heatingDuration = 0;
     // 加热状态
-    public String heatingStatus = "off"; // off:未加热 heating:加热中 holding:加热保持中
-    // 试管存在状态
-    public List<Boolean> tubeExistsMap = new ArrayList<>();
+    public String heatingStatus = "off"; // off:未加热 on:加热中
+    // 试管列表
+    public List<HeatingTurntableSlotTube> tubes = new ArrayList<>();
+
+    // Constructor
+    public HeatingTurntableSlot() {
+        for (int i = 0; i < 16; i++) {
+            tubes.add(null);
+        }
+    }
 
     // 设置试管架编号
     public void setTubeRackNo( String tubeRackNo ) {
         this.tubeRackNo = tubeRackNo;
     }
 
+    // 设置试管
+    public void setTubes( List<HeatingTurntableSlotTube> tubes ) {
+        for (int i = 0; i < tubes.size(); i++) {
+            this.tubes.set(i, tubes.get(i));
+        }
+    }
+
     // 获取非空试管索引列表
     public List<Integer> getExistTubeIndexes() {
         List<Integer> indexes = new ArrayList<>();
-        for (int i = 0; i < tubeExistsMap.size(); i++) {
-            if (tubeExistsMap.get(i)) {
+        for (int i = 0; i < tubes.size(); i++) {
+            if ( null != tubes.get(i) ) {
                 indexes.add(i);
             }
         }
         return indexes;
     }
 
-    public void heating( Integer temperature, Integer duration ) {
     // 执行加热
+    public void heating( Integer temperature, Integer duration ) {
+        this.destTemperature = temperature;
+        this.heatingDuration = duration;
+        this.heatingStatus = "on";
+
+        var snippetName = "HeatingTurntableSlotHeating.Start." + this.index;
+        Map<String,Object> snippetParams = Map.of("temperature", temperature, "duration", duration);
+        UfCmdSnippetExecutor.execute(snippetName, snippetParams);
+
+        // 加热完成
+        this.destTemperature = 0;
+        this.heatingDuration = 0;
+        this.heatingStatus = "off";
     }
 }

src/main/java/com/iflytop/digester/deviceinstance/HeatingTurntableSlotTube.java

@@ -0,0 +1,7 @@
+package com.iflytop.digester.deviceinstance;
+public class HeatingTurntableSlotTube {
+    // tube index
+    public Integer index = -1;
+    // tube no
+    public String no = "";
+}

src/main/java/com/iflytop/digester/deviceinstance/LiquidAdditionInstance.java

@@ -5,7 +5,6 @@ import org.springframework.stereotype.Component;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-
 @Component
 public class LiquidAdditionInstance {
     // list of liquids
@@ -27,43 +26,58 @@ public class LiquidAdditionInstance {
 
     // 针对试管加液
     public void addLiquidToTubes(List<Integer> tubes, String type, int volume ) {
-        var outerIndexes = List.of(0,4, 3,2, 15,11, 12,13);
-        var innerIndexes = List.of(1,5, 6,7, 14,10, 9,8);
         var pumpIndexes = this.getPumpIndexForGroupOutAndIn(type);
         var pumpGroupOutIndex = pumpIndexes.get(0);
         var pumpGroupInIndex = pumpIndexes.get(1);
 
-
-        for ( int i=0; i<4; i++ ) {
-            // @TODO: 这里要计算坐标， 但是还没想好, 等一下 ~~~
-
+        for ( int batchIndex=0; batchIndex<4; batchIndex++ ) {
             // 外圈加液
-            UfCmdSnippetExecutor.execute("LiquidAdditionGroupOutPrepare");
+            UfCmdSnippetExecutor.execute("LiquidAdditionPrepare.Out." + batchIndex);
+            if ( this.checkTubeExists("GOUT", batchIndex, "TOUT", tubes) ) { // 外圈，外侧试管加液
                 UfCmdSnippetExecutor.execute(
                     "LiquidAdditionPump" + pumpGroupOutIndex,
                     Map.of("volume", volume)
                 );
+            }
+            if ( this.checkTubeExists("GOUT", batchIndex, "TIN", tubes) ) { // 外圈，内侧试管加液
                 UfCmdSnippetExecutor.execute(
                     "LiquidAdditionPump" + pumpGroupInIndex,
                     Map.of("volume", volume)
                 );
+            }
 
             // 内圈加液
-            UfCmdSnippetExecutor.execute("LiquidAdditionGroupInPrepare");
+            UfCmdSnippetExecutor.execute("LiquidAdditionPrepare.In." + batchIndex);
+            if ( this.checkTubeExists("GIN", batchIndex, "TOUT", tubes) ) { // 内圈，外侧试管加液
                 UfCmdSnippetExecutor.execute(
                     "LiquidAdditionPump" + pumpGroupOutIndex,
                     Map.of("volume", volume)
                 );
+            }
+            if ( this.checkTubeExists("GIN", batchIndex, "TIN", tubes) ) { // 内圈，内侧试管加液
                 UfCmdSnippetExecutor.execute(
                     "LiquidAdditionPump" + pumpGroupInIndex,
                     Map.of("volume", volume)
                 );
             }
+        }
 
         // 加液复位
         UfCmdSnippetExecutor.execute("LiquidAdditionReset");
     }
 
+    // 检查试管是否存在
+    private Boolean checkTubeExists(String groupType, Integer batchIndex, String slotType, List<Integer> tubes) {
+        var slotIndexes = Map.of(
+            "GIN", List.of(1,5, 6,7, 14,10, 9,8),
+            "GOUT", List.of(0,4, 3,2, 15,11, 12,13)
+        );
+        var groupSlotIndexes = slotIndexes.get(groupType);
+        var batchSlotIndexes = groupSlotIndexes.subList(batchIndex*2, batchIndex*2+2);
+        var slotIndexesToCheck = batchSlotIndexes.get("TOUT".equals(slotType) ? 0 : 1);
+        return tubes.contains(slotIndexesToCheck);
+    }
+
     // 获取蠕动泵索引
     private List<Integer> getPumpIndexForGroupOutAndIn( String type ) {
         for ( int i=0; i<8; i++ ) {
@@ -73,16 +87,4 @@ public class LiquidAdditionInstance {
         }
         throw new RuntimeException("未找到对应的液体类型");
     }
-
-
-    //当前有几种酸
-    //各种酸目前的状态，是否缺液
-    //指挥加酸机械臂运动 ，实际上调用的是 片段
-
-
-
-    //调用蠕动泵加酸，实际上调用的是 片段
-
-
-
 }

src/main/java/com/iflytop/digester/deviceinstance/TransferRobotArmInstance.java

@@ -1,35 +1,86 @@
 package com.iflytop.digester.deviceinstance;
+import com.iflytop.digester.underframework.UfCmdSnippetExecutor;
 import org.springframework.stereotype.Component;
 import java.util.ArrayList;
 import java.util.List;
 @Component
 public class TransferRobotArmInstance {
-
+    // 等待锁队列
     private final List<Object> waitingLocks = new ArrayList<>();
 
+    // 移动试管到试管架
     public void takeOutTubesFromErrorSlot(List<Integer> tubeIndexes) {
+        var lock = new Object();
+        this.lockArm(lock);
+
         // 执行从异常槽位取出试管
+        UfCmdSnippetExecutor.execute("TubeTakeOutFromErrorSlotStart");
+        for (Integer tubeIndex : tubeIndexes) {
+            UfCmdSnippetExecutor.execute("TubeTakeOutFromErrorSlot." + tubeIndex);
         }
+        UfCmdSnippetExecutor.execute("TubeTakeOutFromErrorSlotEnd");
 
+        this.unlockArm(lock);
+    }
+
+    // 移动试管到异常槽位
     public void moveTubesToErrorSlot(List<Integer> tubeIndexes) {
+        var lock = new Object();
+        this.lockArm(lock);
+
         // 执行移动试管到异常槽位
+        UfCmdSnippetExecutor.execute("TubeMoveToErrorSlotStart");
+        for (Integer tubeIndex : tubeIndexes) {
+            UfCmdSnippetExecutor.execute("TubeMoveToErrorSlot." + tubeIndex);
+        }
+        UfCmdSnippetExecutor.execute("TubeMoveToErrorSlotEnd");
+
+        this.unlockArm(lock);
     }
 
+    // 移动试管架到加液盘
     public void moveTubeRackToLiquidPlate(Integer srcSlotIndex) {
-        // 执行移动试管架到液体盘
+        var lock = new Object();
+        this.lockArm(lock);
+
+        // 执行移动试管架到加液盘
+        var snippetName = "TubeRackMoveToLiquidPlate." + srcSlotIndex;
+        UfCmdSnippetExecutor.execute(snippetName);
+
+        this.unlockArm(lock);
     }
 
+    // 移动试管架到加热转盘
     public void moveTubeRackToHeatingTurntable(Integer slot) throws InterruptedException {
-        if ( !waitingLocks.isEmpty() ) {
         var lock = new Object();
+        this.lockArm(lock);
+
+        // 执行移动试管架到加热转盘
+        UfCmdSnippetExecutor.execute("TubeRackMoveToHeatingTurntable." + slot);
+
+        this.unlockArm(lock);
+    }
+
+    // wait for arm
+    private void lockArm( Object lock ) {
+        if ( waitingLocks.isEmpty() ) {
+            this.waitingLocks.add(lock);
+            return ;
+        }
+
+        try {
             waitingLocks.add(lock);
             lock.wait();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
     }
 
-        // 执行移动试管架到加热转盘
-
+    // unlock arm
+    private void unlockArm( Object lock ) {
+        waitingLocks.remove(lock);
         if ( !waitingLocks.isEmpty() ) {
-            waitingLocks.remove(0).notify();
+            waitingLocks.get(0).notify();
         }
     }
 }

src/main/java/com/iflytop/digester/model/MdbDigestionTask.java

@@ -1,6 +1,12 @@
 package com.iflytop.digester.model;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iflytop.digester.deviceinstance.HeatingTurntableSlotTube;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecord;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecordField;
+import java.util.ArrayList;
+import java.util.List;
 public class MdbDigestionTask extends UfActiveRecord {
     @UfActiveRecordField
     public String digestionId;
@@ -29,8 +35,33 @@ public class MdbDigestionTask extends UfActiveRecord {
     @UfActiveRecordField
     public Integer finishedAt = 0;
 
+    @UfActiveRecordField
+    public Integer heatingSlotIndex = -1;
+
     // get table name
     public static String getTableName() {
         return "app_digestion_tasks";
     }
+
+    // get tubes
+    public List<HeatingTurntableSlotTube> getTubes() {
+        List<HeatingTurntableSlotTube> list = new ArrayList<>();
+
+        ObjectMapper jsonMapper = new ObjectMapper();
+        JsonNode tubesJsonTree = null;
+        try {
+            tubesJsonTree = jsonMapper.readTree(this.tubes);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+
+        var indexCounter = 0;
+        for ( JsonNode tubeNode : tubesJsonTree ) {
+            var tube = new HeatingTurntableSlotTube();
+            tube.index = indexCounter++;
+            tube.no = tubeNode.get("no").asText();
+            list.add(tube);
+        }
+        return list;
+    }
 }

src/main/java/com/iflytop/digester/underframework/UfActuatorCmdExecutor.java

@@ -0,0 +1,7 @@
+package com.iflytop.digester.underframework;
+import com.iflytop.digester.underframework.dao.model.UfMdbActuatorCmd;
+public class UfActuatorCmdExecutor {
+    public static void execute(UfMdbActuatorCmd cmd) {
+        // @TODO : 这里要判断使用那种通讯方式进行执行， 所以待定一下
+    }
+}

src/main/java/com/iflytop/digester/underframework/UfCmdSnippetExecutor.java

@@ -1,12 +1,30 @@
 package com.iflytop.digester.underframework;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iflytop.digester.deviceinstance.HeatingTurntableSlotTube;
+import com.iflytop.digester.underframework.dao.model.UfMdbActuatorCmd;
+import com.iflytop.digester.underframework.dao.model.UfMdbSnippet;
+import com.iflytop.digester.underframework.dao.record.UfActiveRecord;
+import com.iflytop.digester.underframework.util.UfJsonHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.HashMap;
 import java.util.Map;
 public class UfCmdSnippetExecutor {
+    // logger
+    public static final Logger LOG = LoggerFactory.getLogger(UfCmdSnippetExecutor.class);
+    // snippet
+    private final UfMdbSnippet snippet;
+    // params
+    private final Map<String, Object> params;
+
     /**
      * execute snippet
      * @param snippetName - snippet name
      */
     static public void execute(String snippetName) {
-
+        UfCmdSnippetExecutor.execute(snippetName, new HashMap<>());
     }
 
     /**
@@ -15,6 +33,42 @@ public class UfCmdSnippetExecutor {
      * @param params - params
      */
     static public void execute(String snippetName, Map<String, Object> params) {
+        LOG.info("[Snippet Execute] : {} ({})", snippetName, UfJsonHelper.objectToJson(params));
+
+//        var snippet = UfActiveRecord.findOne(UfMdbSnippet.class, Map.of("key", snippetName));
+//        if (null == snippet) {
+//            throw new RuntimeException("无效的片段名称 : " + snippetName);
+//        }
+//
+//        var executor = new UfCmdSnippetExecutor(snippet, params);
+//        executor.run();
+    }
+
+    // constructor
+    public UfCmdSnippetExecutor( UfMdbSnippet snippet, Map<String, Object> params ) {
+        this.snippet = snippet;
+        this.params = params;
+    }
 
+    // execute snippet
+    public void run() {
+        ObjectMapper jsonMapper = new ObjectMapper();
+        JsonNode cmdsJsonTree = null;
+        try {
+            cmdsJsonTree = jsonMapper.readTree(this.snippet.cmds);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+
+        for ( JsonNode cmdNode : cmdsJsonTree ) {
+            var cmdId = cmdNode.get("cmdId").asText();
+            var cmd = UfActiveRecord.findOne(UfMdbActuatorCmd.class, cmdId);
+            if (null == cmd) {
+                throw new RuntimeException("无效的指令ID : " + cmdId);
+            }
+
+            cmd.parameters = cmdNode.get("params").asText();
+            UfActuatorCmdExecutor.execute(cmd);
+        }
     }
 }

src/main/java/com/iflytop/digester/underframework/connection/UfZcancmderWebsocket.java

@@ -1,6 +1,6 @@
 package com.iflytop.digester.underframework.connection;
 import com.iflytop.digester.underframework.dao.model.TsMdbActuator;
-import com.iflytop.digester.underframework.dao.model.TsMdbActuatorCmd;
+import com.iflytop.digester.underframework.dao.model.UfMdbActuatorCmd;
 import com.iflytop.digester.underframework.util.TsByteBuffer;
 import org.java_websocket.client.WebSocketClient;
 import org.java_websocket.handshake.ServerHandshake;
@@ -62,7 +62,7 @@ public class UfZcancmderWebsocket extends UfConnectionBase {
     }
 
     // execute command
-    synchronized public void execute( TsMdbActuatorCmd actuatorCmd ) {
+    synchronized public void execute( UfMdbActuatorCmd actuatorCmd ) {
         this.executeDeviceCommand(actuatorCmd);
         if ( 1 == actuatorCmd.waitForFinish ) {
             this.waitForActuatorFinish(actuatorCmd);
@@ -70,9 +70,9 @@ public class UfZcancmderWebsocket extends UfConnectionBase {
     }
 
     // wait for actuator finish
-    private void waitForActuatorFinish(TsMdbActuatorCmd actuatorCmd) {
+    private void waitForActuatorFinish(UfMdbActuatorCmd actuatorCmd) {
         do {
-            var waitCmd = new TsMdbActuatorCmd();
+            var waitCmd = new UfMdbActuatorCmd();
             waitCmd.actuatorId = actuatorCmd.actuatorId;
             waitCmd.cmdId = "0104";
             waitCmd.cmdKey = "module_get_status";
@@ -92,7 +92,7 @@ public class UfZcancmderWebsocket extends UfConnectionBase {
     }
 
     // execute device command
-    private void executeDeviceCommand( TsMdbActuatorCmd actuatorCmd ) {
+    private void executeDeviceCommand( UfMdbActuatorCmd actuatorCmd ) {
         this.response = null;
         this.responseError = null;
         this.sendCommandRequest(actuatorCmd);
@@ -111,7 +111,7 @@ public class UfZcancmderWebsocket extends UfConnectionBase {
     }
 
     // send command request
-    private void sendCommandRequest(TsMdbActuatorCmd actuatorCmd) {
+    private void sendCommandRequest(UfMdbActuatorCmd actuatorCmd) {
         String cmd = this.buildCommand(actuatorCmd);
 
         TsMdbActuator actuator = TsMdbActuator.findOne(TsMdbActuator.class, actuatorCmd.actuatorId);
@@ -131,7 +131,7 @@ public class UfZcancmderWebsocket extends UfConnectionBase {
     }
 
     // build command
-    private String buildCommand(TsMdbActuatorCmd actuatorCmd) {
+    private String buildCommand(UfMdbActuatorCmd actuatorCmd) {
         String params = actuatorCmd.parameters.trim();
         List<String> paramList = new ArrayList<>();
         if ( !params.isEmpty() ) {

src/main/java/com/iflytop/digester/underframework/dao/model/TsMdbActuatorCmd.java → src/main/java/com/iflytop/digester/underframework/dao/model/UfMdbActuatorCmd.java

@@ -1,7 +1,7 @@
 package com.iflytop.digester.underframework.dao.model;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecord;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecordField;
-public class TsMdbActuatorCmd extends UfActiveRecord {
+public class UfMdbActuatorCmd extends UfActiveRecord {
     @UfActiveRecordField
     public String actuatorId;
 

src/main/java/com/iflytop/digester/underframework/dao/model/TsMdbSnippet.java → src/main/java/com/iflytop/digester/underframework/dao/model/UfMdbSnippet.java

@@ -5,7 +5,7 @@ import com.iflytop.digester.underframework.dao.record.UfActiveRecordField;
 /**
  * 执行片段
  */
-public class TsMdbSnippet extends UfActiveRecord {
+public class UfMdbSnippet extends UfActiveRecord {
     @UfActiveRecordField
     public String name;
 

src/main/java/com/iflytop/digester/underframework/web/api/TsApiActuatorCommand.java

@@ -1,7 +1,7 @@
 package com.iflytop.digester.underframework.web.api;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecord;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecordCriteria;
-import com.iflytop.digester.underframework.dao.model.TsMdbActuatorCmd;
+import com.iflytop.digester.underframework.dao.model.UfMdbActuatorCmd;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -15,13 +15,13 @@ public class TsApiActuatorCommand extends UfApiControllerBase {
         String actuatorId = (String) params.get("actuatorId");
         var criteria = new UfActiveRecordCriteria();
         criteria.conditions = Map.of("actuatorId", actuatorId);
-        var items = UfActiveRecord.find(TsMdbActuatorCmd.class, criteria);
+        var items = UfActiveRecord.find(UfMdbActuatorCmd.class, criteria);
         return success(items);
     }
 
     @PostMapping("/api/actuator-cmd/save")
     @ResponseBody
-    public UfApiResponse save(@RequestBody TsMdbActuatorCmd actuatorCmd ) {
+    public UfApiResponse save(@RequestBody UfMdbActuatorCmd actuatorCmd ) {
         if ( null != actuatorCmd.id ) {
             actuatorCmd.isNewRecord = false;
         }
@@ -31,14 +31,14 @@ public class TsApiActuatorCommand extends UfApiControllerBase {
 
     @PostMapping("/api/actuator-cmd/delete")
     @ResponseBody
-    public UfApiResponse delete(@RequestBody TsMdbActuatorCmd actuatorCmd ) {
+    public UfApiResponse delete(@RequestBody UfMdbActuatorCmd actuatorCmd ) {
         actuatorCmd.delete();
         return success();
     }
 
     @PostMapping("/api/actuator-cmd/execute")
     @ResponseBody
-    public UfApiResponse execute(@RequestBody TsMdbActuatorCmd actuatorCmd ) {
+    public UfApiResponse execute(@RequestBody UfMdbActuatorCmd actuatorCmd ) {
         try {
 //            UfApplication.getApp().commandExecutor.execute(actuatorCmd);
         } catch ( Exception e ) {

src/main/java/com/iflytop/digester/underframework/web/api/TsApiSnippet.java

@@ -1,6 +1,6 @@
 package com.iflytop.digester.underframework.web.api;
 import com.iflytop.digester.underframework.dao.record.UfActiveRecord;
-import com.iflytop.digester.underframework.dao.model.TsMdbSnippet;
+import com.iflytop.digester.underframework.dao.model.UfMdbSnippet;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -11,13 +11,13 @@ public class TsApiSnippet extends UfApiControllerBase {
     @PostMapping("/api/snippet/list")
     @ResponseBody
     public UfApiResponse list() {
-        var items = UfActiveRecord.find(TsMdbSnippet.class);
+        var items = UfActiveRecord.find(UfMdbSnippet.class);
         return success(items);
     }
 
     @PostMapping("/api/snippet/save")
     @ResponseBody
-    public UfApiResponse save(@RequestBody TsMdbSnippet snippet ) {
+    public UfApiResponse save(@RequestBody UfMdbSnippet snippet ) {
         if ( null != snippet.id ) {
             snippet.isNewRecord = false;
         }
@@ -27,16 +27,16 @@ public class TsApiSnippet extends UfApiControllerBase {
 
     @PostMapping("/api/snippet/delete")
     @ResponseBody
-    public UfApiResponse delete(@RequestBody TsMdbSnippet snippet ) {
+    public UfApiResponse delete(@RequestBody UfMdbSnippet snippet ) {
         snippet.delete();
         return success();
     }
 
     @PostMapping("/api/snippet/execute")
     @ResponseBody
-    public UfApiResponse execute(@RequestBody TsMdbSnippet snippet ) {
+    public UfApiResponse execute(@RequestBody UfMdbSnippet snippet ) {
         try {
-            snippet = UfActiveRecord.findOne(TsMdbSnippet.class, snippet.id);
+            snippet = UfActiveRecord.findOne(UfMdbSnippet.class, snippet.id);
 //            TsSnippetExecutor.executeSnippet(snippet);
         } catch ( Exception e ) {
             return error(e.getMessage());

src/main/resources/application.yml

@@ -13,3 +13,12 @@ device:
     - name : zcancmder
       type : ZcancmderWebsocket
       uri: ws://192.168.8.10:19005
+
+mqtt-broker:
+  uri: tcp://broker.emqx.io:1883
+  username: admin
+  password: public
+  clientId: digester
+  my-topic : "stw-a80"
+  transbot-topic : "transbot"
+

web

@@ -1 +1 @@
-Subproject commit bb327e921950cf4cfad799a02f6c7faa076919f4
+Subproject commit 92cb587baccf83f33dd9465bb37dc815058f26a0