diff --git a/scada-common/rsio.lua b/scada-common/rsio.lua index d71d777..92e1713 100644 --- a/scada-common/rsio.lua +++ b/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 diff --git a/supervisor/session/rtu.lua b/supervisor/session/rtu.lua index 5dab40c..fab3a96 100644 --- a/supervisor/session/rtu.lua +++ b/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 - local _parse_advertisement = function () + -- parse the recorded advertisement and create unit sub-sessions + 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], - index = self.advert[i][1], - reactor = self.advert[i][2], - rsio = self.advert[i][3] + type = self.advert[i][1], + index = self.advert[i][2], + reactor = 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 -- ---------------------- diff --git a/supervisor/session/rtu/boiler.lua b/supervisor/session/rtu/boiler.lua index 320d7e4..c0dd0e6 100644 --- a/supervisor/session/rtu/boiler.lua +++ b/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 diff --git a/supervisor/session/rtu/emachine.lua b/supervisor/session/rtu/emachine.lua index 9428806..cdbf21e 100644 --- a/supervisor/session/rtu/emachine.lua +++ b/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 diff --git a/supervisor/session/rtu/redstone.lua b/supervisor/session/rtu/redstone.lua new file mode 100644 index 0000000..1e1c1b9 --- /dev/null +++ b/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 diff --git a/supervisor/session/rtu/turbine.lua b/supervisor/session/rtu/turbine.lua index 66429f3..292ff95 100644 --- a/supervisor/session/rtu/turbine.lua +++ b/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