#8 bugfixes, redstone RTU session

scada-common/rsio.lua

@@ -11,7 +11,8 @@ local rsio = {}
 ---@alias IO_LVL integer
 local IO_LVL = {
     LOW = 0,
-    HIGH = 1
+    HIGH = 1,
+    DISCONNECT = 2  -- use for RTU session to indicate this RTU is not connected to this channel
 }
 
 ---@alias IO_DIR integer

supervisor/session/rtu.lua

@@ -6,6 +6,7 @@ local util = require("scada-common.util")
 -- supervisor rtu sessions (svrs)
 local svrs_boiler = require("supervisor.session.rtu.boiler")
 local svrs_emachine = require("supervisor.session.rtu.emachine")
+local svrs_redstone = require("supervisor.session.rtu.redstone")
 local svrs_turbine = require("supervisor.session.rtu.turbine")
 
 local rtu = {}
@@ -19,10 +20,26 @@ local println = util.println
 local print_ts = util.print_ts
 local println_ts = util.println_ts
 
+local RTU_S_CMDS = {
+}
+
+local RTU_S_DATA = {
+    RS_COMMAND = 1,
+    UNIT_COMMAND = 2
+}
+
+rtu.RTU_S_CMDS = RTU_S_CMDS
+rtu.RTU_S_DATA = RTU_S_DATA
+
 local PERIODICS = {
     KEEP_ALIVE = 2.0
 }
 
+---@class rs_session_command
+---@field reactor integer
+---@field channel RS_IO
+---@field active boolean
+
 -- create a new RTU session
 ---@param id integer
 ---@param in_queue mqueue
@@ -42,31 +59,32 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
         connected = true,
         rtu_conn_watchdog = util.new_watchdog(3),
         last_rtt = 0,
+        rs_io = {},
         units = {}
     }
 
     ---@class rtu_session
     local public = {}
 
-    -- parse the recorded advertisement
+    -- parse the recorded advertisement and create unit sub-sessions
-    local _parse_advertisement = function ()
+    local _handle_advertisement = function ()
         self.units = {}
         for i = 1, #self.advert do
-            local unit = nil    ---@type unit_session
+            local unit = nil    ---@type unit_session|nil
 
             ---@type rtu_advertisement
             local unit_advert = {
-                type = self.advert[i][0],
+                type = self.advert[i][1],
-                index = self.advert[i][1],
+                index = self.advert[i][2],
-                reactor = self.advert[i][2],
+                reactor = self.advert[i][3],
-                rsio = self.advert[i][3]
+                rsio = self.advert[i][4]
             }
 
             local u_type = unit_advert.type
 
             -- create unit by type
             if u_type == RTU_UNIT_TYPES.REDSTONE then
-
+                unit = svrs_redstone.new(self.id, unit_advert, self.out_q)
             elseif u_type == RTU_UNIT_TYPES.BOILER then
                 unit = svrs_boiler.new(self.id, unit_advert, self.out_q)
             elseif u_type == RTU_UNIT_TYPES.BOILER_VALVE then
@@ -79,27 +97,20 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
                 unit = svrs_emachine.new(self.id, unit_advert, self.out_q)
             elseif u_type == RTU_UNIT_TYPES.IMATRIX then
                 -- @todo Mekanism 10.1+
+            else
+                log.error(log_header .. "bad advertisement: encountered unsupported RTU type")
             end
 
             if unit ~= nil then
                 table.insert(self.units, unit)
             else
                 self.units = {}
-                log.error(log_header .. "bad advertisement; encountered unsupported RTU type")
+                log.error(log_header .. "bad advertisement: error occured while creating a unit")
                 break
             end
         end
     end
 
-    -- send a MODBUS TCP packet
-    ---@param m_pkt modbus_packet
-    local _send_modbus = function (m_pkt)
-        local s_pkt = comms.scada_packet()
-        s_pkt.make(self.seq_num, PROTOCOLS.MODBUS_TCP, m_pkt.raw_sendable())
-        self.modem.transmit(self.s_port, self.l_port, s_pkt.raw_sendable())
-        self.seq_num = self.seq_num + 1
-    end
-
     -- send a SCADA management packet
     ---@param msg_type SCADA_MGMT_TYPES
     ---@param msg table
@@ -137,7 +148,7 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
                 unit.handle_packet(pkt)
             end
         elseif pkt.scada_frame.protocol() == PROTOCOLS.SCADA_MGMT then
-
+            -- handle management packet
             if pkt.type == SCADA_MGMT_TYPES.KEEP_ALIVE then
                 -- keep alive reply
                 if pkt.length == 2 then
@@ -160,8 +171,9 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
                 self.connected = false
             elseif pkt.type == SCADA_MGMT_TYPES.RTU_ADVERT then
                 -- RTU unit advertisement
+                -- handle advertisement; this will re-create all unit sub-sessions
                 self.advert = pkt.data
-                _parse_advertisement()
+                _handle_advertisement()
             else
                 log.debug(log_header .. "handler received unsupported SCADA_MGMT packet type " .. pkt.type)
             end
@@ -192,14 +204,6 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
     ---@return boolean connected
     public.iterate = function ()
         if self.connected then
-            ------------------
-            -- update units --
-            ------------------
-
-            for i = 1, #self.units do
-                self.units[i].update()
-            end
-
             ------------------
             -- handle queue --
             ------------------
@@ -218,6 +222,11 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
                         -- handle instruction
                     elseif msg.qtype == mqueue.TYPE.DATA then
                         -- instruction with body
+                        local cmd = msg.message ---@type queue_data
+
+                        if cmd.key == RTU_S_DATA.RS_COMMAND then
+                            local rs_cmd = cmd.val  ---@type rs_session_command
+                        end
                     end
                 end
 
@@ -236,6 +245,14 @@ rtu.new_session = function (id, in_queue, out_queue, advertisement)
                 return self.connected
             end
 
+            ------------------
+            -- update units --
+            ------------------
+
+            for i = 1, #self.units do
+                self.units[i].update()
+            end
+
             ----------------------
             -- update periodics --
             ----------------------

supervisor/session/rtu/boiler.lua

@@ -183,19 +183,19 @@ boiler.new = function (session_id, advert, out_queue)
     -- update this runner
     ---@param time_now integer milliseconds
     public.update = function (time_now)
-        if not self.has_build and self.next_build_req <= time_now then
+        if not self.periodics.has_build and self.next_build_req <= time_now then
             _request_build()
-            self.next_build_req = time_now + PERIODICS.BUILD
+            self.periodics.next_build_req = time_now + PERIODICS.BUILD
         end
 
-        if self.next_state_req <= time_now then
+        if self.periodics.next_state_req <= time_now then
             _request_state()
-            self.next_state_req = time_now + PERIODICS.STATE
+            self.periodics.next_state_req = time_now + PERIODICS.STATE
         end
 
-        if self.next_tanks_req <= time_now then
+        if self.periodics.next_tanks_req <= time_now then
             _request_tanks()
-            self.next_tanks_req = time_now + PERIODICS.TANKS
+            self.periodics.next_tanks_req = time_now + PERIODICS.TANKS
         end
     end
 

supervisor/session/rtu/emachine.lua

@@ -133,14 +133,14 @@ emachine.new = function (session_id, advert, out_queue)
     -- update this runner
     ---@param time_now integer milliseconds
     public.update = function (time_now)
-        if not self.has_build and self.next_build_req <= time_now then
+        if not self.has_build and self.periodics.next_build_req <= time_now then
             _request_build()
-            self.next_build_req = time_now + PERIODICS.BUILD
+            self.periodics.next_build_req = time_now + PERIODICS.BUILD
         end
 
-        if self.next_storage_req <= time_now then
+        if self.periodics.next_storage_req <= time_now then
             _request_storage()
-            self.next_storage_req = time_now + PERIODICS.STORAGE
+            self.periodics.next_storage_req = time_now + PERIODICS.STORAGE
         end
     end
 

supervisor/session/rtu/redstone.lua

@@ -0,0 +1,183 @@
+local comms = require("scada-common.comms")
+local log = require("scada-common.log")
+local rsio = require("scada-common.rsio")
+local types = require("scada-common.types")
+
+local txnctrl = require("supervisor.session.rtu.txnctrl")
+
+local redstone = {}
+
+local PROTOCOLS = comms.PROTOCOLS
+local RTU_UNIT_TYPES = comms.RTU_UNIT_TYPES
+local MODBUS_FCODE = types.MODBUS_FCODE
+
+local RS_IO = rsio.IO
+local IO_LVL = rsio.IO_LVL
+local IO_DIR = rsio.IO_DIR
+local IO_MODE = rsio.IO_MODE
+
+local rtu_t = types.rtu_t
+
+local TXN_TYPES = {
+    DI_READ = 0,
+    INPUT_REG_READ = 1
+}
+
+local PERIODICS = {
+    INPUT_READ = 200
+}
+
+-- create a new redstone rtu session runner
+---@param session_id integer
+---@param advert rtu_advertisement
+---@param out_queue mqueue
+redstone.new = function (session_id, advert, out_queue)
+    -- type check
+    if advert.type ~= RTU_UNIT_TYPES.REDSTONE then
+        log.error("attempt to instantiate redstone RTU for type '" .. advert.type .. "'. this is a bug.")
+        return nil
+    end
+
+    local log_tag = "session.rtu(" .. session_id .. ").redstone(" .. advert.index .. "): "
+
+    local self = {
+        uid = advert.index,
+        reactor = advert.reactor,
+        out_q = out_queue,
+        transaction_controller = txnctrl.new(),
+        has_di = false,
+        has_ai = false,
+        periodics = {
+            next_di_req = 0,
+            next_ir_req = 0,
+        },
+        io_list = {
+            digital_in = {},    -- discrete inputs
+            digital_out = {},   -- coils
+            analog_in = {},     -- input registers
+            analog_out = {}     -- holding registers
+        },
+        db = {}
+    }
+
+    ---@class unit_session
+    local public = {}
+
+    -- INITIALIZE --
+
+    for _ = 1, #RS_IO do
+        table.insert(self.db, IO_LVL.DISCONNECT)
+    end
+
+    for i = 1, #advert.rsio do
+        local channel = advert.rsio[i]
+        local mode = rsio.get_io_mode(channel)
+
+        if mode == IO_MODE.DIGITAL_IN then
+            self.has_di = true
+            table.insert(self.io_list.digital_in, channel)
+        elseif mode == IO_MODE.DIGITAL_OUT then
+            table.insert(self.io_list.digital_out, channel)
+        elseif mode == IO_MODE.ANALOG_IN then
+            self.has_ai = true
+            table.insert(self.io_list.analog_in, channel)
+        elseif mode == IO_MODE.ANALOG_OUT then
+            table.insert(self.io_list.analog_out, channel)
+        else
+            -- should be unreachable code, we already validated channels
+            log.error(log_tag .. "failed to identify advertisement channel IO mode (" .. channel .. ")", true)
+            return nil
+        end
+
+        self.db[channel] = IO_LVL.LOW
+    end
+
+
+    -- PRIVATE FUNCTIONS --
+
+    local _send_request = function (txn_type, f_code, register_range)
+        local m_pkt = comms.modbus_packet()
+        local txn_id = self.transaction_controller.create(txn_type)
+
+        m_pkt.make(txn_id, self.uid, f_code, register_range)
+
+        self.out_q.push_packet(m_pkt)
+    end
+
+    -- query discrete inputs
+    local _request_discrete_inputs = function ()
+        _send_request(TXN_TYPES.DI_READ, MODBUS_FCODE.READ_DISCRETE_INPUTS, { 1, #self.io_list.digital_in })
+    end
+
+    -- query input registers
+    local _request_input_registers = function ()
+        _send_request(TXN_TYPES.INPUT_REG_READ, MODBUS_FCODE.READ_INPUT_REGS, { 1, #self.io_list.analog_in })
+    end
+
+    -- PUBLIC FUNCTIONS --
+
+    -- handle a packet
+    ---@param m_pkt modbus_frame
+    public.handle_packet = function (m_pkt)
+        local success = false
+
+        if m_pkt.scada_frame.protocol() == PROTOCOLS.MODBUS_TCP then
+            if m_pkt.unit_id == self.uid then
+                local txn_type = self.transaction_controller.resolve(m_pkt.txn_id)
+                if txn_type == TXN_TYPES.DI_READ then
+                    -- build response
+                    if m_pkt.length == 1 then
+                        self.db.build.max_energy = m_pkt.data[1]
+                    else
+                        log.debug(log_tag .. "MODBUS transaction reply length mismatch (emachine.build)")
+                    end
+                elseif txn_type == TXN_TYPES.INPUT_REG_READ then
+                    -- storage response
+                    if m_pkt.length == 3 then
+                        self.db.storage.energy = m_pkt.data[1]
+                        self.db.storage.energy_need = m_pkt.data[2]
+                        self.db.storage.energy_fill = m_pkt.data[3]
+                    else
+                        log.debug(log_tag .. "MODBUS transaction reply length mismatch (emachine.storage)")
+                    end
+                elseif txn_type == nil then
+                    log.error(log_tag .. "unknown transaction reply")
+                else
+                    log.error(log_tag .. "unknown transaction type " .. txn_type)
+                end
+            else
+                log.error(log_tag .. "wrong unit ID: " .. m_pkt.unit_id, true)
+            end
+        else
+            log.error(log_tag .. "illegal packet type " .. m_pkt.scada_frame.protocol(), true)
+        end
+
+        return success
+    end
+
+    public.get_uid = function () return self.uid end
+    public.get_reactor = function () return self.reactor end
+    public.get_db = function () return self.db end
+
+    -- update this runner
+    ---@param time_now integer milliseconds
+    public.update = function (time_now)
+        if self.has_di then
+            if self.periodics.next_di_req <= time_now then
+                _request_discrete_inputs()
+                self.periodics.next_di_req = time_now + PERIODICS.BUILD
+            end
+        end
+
+        if self.has_ai then
+            if self.periodics.next_ir_req <= time_now then
+                _request_input_registers()
+                self.periodics.next_ir_req = time_now + PERIODICS.STORAGE
+            end
+        end
+    end
+
+    return public
+end
+
+return redstone

supervisor/session/rtu/turbine.lua

@@ -174,19 +174,19 @@ turbine.new = function (session_id, advert, out_queue)
     -- update this runner
     ---@param time_now integer milliseconds
     public.update = function (time_now)
-        if not self.has_build and self.next_build_req <= time_now then
+        if not self.has_build and self.periodics.next_build_req <= time_now then
             _request_build()
-            self.next_build_req = time_now + PERIODICS.BUILD
+            self.periodics.next_build_req = time_now + PERIODICS.BUILD
         end
 
-        if self.next_state_req <= time_now then
+        if self.periodics.next_state_req <= time_now then
             _request_state()
-            self.next_state_req = time_now + PERIODICS.STATE
+            self.periodics.next_state_req = time_now + PERIODICS.STATE
         end
 
-        if self.next_tanks_req <= time_now then
+        if self.periodics.next_tanks_req <= time_now then
             _request_tanks()
-            self.next_tanks_req = time_now + PERIODICS.TANKS
+            self.periodics.next_tanks_req = time_now + PERIODICS.TANKS
         end
     end