Merge pull request #39 from joyja/bidirectional

Bidirectional
joyja · Mar 27, 2020 · b0535ea · b0535ea
2 parents ec4aa23 + 7cfb62d
commit b0535ea
Show file tree

Hide file tree

Showing 5 changed files with 93 additions and 7 deletions.
diff --git a/src/device/ethernetip.js b/src/device/ethernetip.js
@@ -100,7 +100,12 @@ class EthernetIPSource extends Model {
   async read() {
     if (this.ethernetip.connected) {
       await this.ethernetip.client.readTag(this.tagData)
-      await this.tag.setValue(this.tagData.value)
+      await this.tag.setValue(this.tagData.value, false)
+    }
+  }
+  async write(value) {
+    if (this.ethernetip.connected) {
+      await this.ethernetip.client.writeTag(this.tagData, value)
     }
   }
   get tagname() {

diff --git a/src/device/modbus.js b/src/device/modbus.js
@@ -216,12 +216,12 @@ class ModbusSource extends Model {
     const view = new DataView(buffer)
     let value = null
     if (this.tag.datatype === `FLOAT`) {
-      view.setInt16(
+      view.setUint16(
         0,
         this.modbus.reverseWords ? data[1] : data[0],
         this.modbus.reverseBits
       )
-      view.setInt16(
+      view.setUint16(
         2,
         this.modbus.reverseWords ? data[0] : data[1],
         this.modbus.reverseBits
@@ -238,12 +238,48 @@ class ModbusSource extends Model {
         this.modbus.reverseWords ? data[0] : data[1],
         this.modbus.reverseBits
       )
+      value = view.getInt32(0, this.modbus.reverseBits)
     } else if (this.tag.datatype === `INT16`) {
       view.setInt16(0, data[0], this.modbus.reverseBits)
       value = view.getInt16(0, this.modbus.reverseBits)
     }
     return value
   }
+  formatOutput(value) {
+    const buffer = new ArrayBuffer(4)
+    const view = new DataView(buffer)
+    let data = []
+    if (this.tag.datatype === `FLOAT`) {
+      view.setFloat32(0, value)
+      data.push(
+        view.getUint16(
+          this.modbus.reverseWords ? 2 : 0,
+          this.modbus.reverseBits
+        )
+      )
+      data.push(
+        view.getUint16(
+          this.modbus.reverseWords ? 0 : 2,
+          this.modbus.reverseBits
+        )
+      )
+    } else if (this.tag.datatype === `INT32`) {
+      view.setInt32(0, value)
+      data.push(
+        view.getUint16(
+          this.modbus.reverseWords ? 0 : 2,
+          !this.modbus.reverseBits
+        )
+      )
+      data.push(
+        view.getUint16(
+          this.modbus.reverseWords ? 2 : 0,
+          !this.modbus.reverseBits
+        )
+      )
+    }
+    return data
+  }
   async read() {
     if (this.modbus.connected) {
       if (this.registerType === 'INPUT_REGISTER') {
@@ -258,7 +294,7 @@ class ModbusSource extends Model {
                 return
               }
               if (data) {
-                await this.tag.setValue(this.formatValue(data.data))
+                await this.tag.setValue(this.formatValue(data.data), false)
               }
               resolve()
             }
@@ -281,7 +317,7 @@ class ModbusSource extends Model {
                 return
               }
               if (data) {
-                this.tag.setValue(this.formatValue(data.data))
+                this.tag.setValue(this.formatValue(data.data), false)
               }
               resolve()
             }
@@ -303,7 +339,7 @@ class ModbusSource extends Model {
                 return
               } else {
                 if (data) {
-                  await this.tag.setValue(data.data[0])
+                  await this.tag.setValue(data.data[0], false)
                 }
                 resolve()
               }
@@ -318,6 +354,39 @@ class ModbusSource extends Model {
       }
     }
   }
+  async write(value) {
+    if (this.modbus.connected) {
+      if (this.registerType === 'HOLDING_REGISTER') {
+        return new Promise((resolve, reject) => {
+          this.modbus.client.writeRegisters(
+            this.register,
+            this.formatOutput(value),
+            (error) => {
+              if (error) {
+                reject(error)
+                return
+              }
+              resolve()
+            }
+          )
+        }).catch(async (error) => {
+          if (error.name === 'TransactionTimedOutError') {
+            await this.modbus.disconnect()
+            await this.modbus.connect()
+          } else {
+            throw error
+          }
+        })
+      } else if (this.registerType === 'COIL') {
+        return new Promise((resolve, reject) => {
+          this.modbus.client.writeCoil(
+            this.register,
+            this.value + '' === 'true'
+          )
+        })
+      }
+    }
+  }
   get register() {
     this.checkInit()
     return this._register

diff --git a/src/relations.js b/src/relations.js
@@ -357,6 +357,14 @@ Mqtt.prototype.publishHistory = async function() {
   await this.constructor.executeUpdate(sql)
 }
 
+Mqtt.prototype.onDcmd = function(payload) {
+  const { metrics } = payload
+  for (metric of metrics) {
+    const tag = Tag.instances.find((tag) => metric.name === tag.name)
+    tag.setValue(metric.value)
+  }
+}
+
 Object.defineProperties(Mqtt.prototype, {
   service: {
     get() {

diff --git a/src/service/mqtt.js b/src/service/mqtt.js
@@ -132,6 +132,7 @@ class Mqtt extends Model {
         logger.info(
           `Mqtt service ${this.service.name} received a dcmd for ${deviceId}.`
         )
+        this.onDcmd(payload)
       })
       this.client.on('ncmd', (payload) => {
         if (payload.metrics) {

diff --git a/src/tag.js b/src/tag.js
@@ -86,8 +86,11 @@ class Tag extends Model {
       return this._value
     }
   }
-  setValue(value) {
+  async setValue(value, write = true) {
     this.checkInit()
+    if (this.source && write) {
+      this.source.write(value)
+    }
     return this.update(this.id, 'value', value, Tag).then((result) => {
       this._value = result
       this.pubsub.publish('tagUpdate', { tagUpdate: this })