GitHub/cc-scada
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#134 #104 redstone RTU integration with supervisor unit, waste routing implemented, changed how redstone I/O works (again, should be good now), modbus fixes

Browse Source
This commit is contained in:
Mikayla Fischler
2022-11-30 23:31:14 -05:00
parent 9c27ac7ae6
commit e1d7c7b1c0
17 changed files with 527 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
supervisor/session/rtu/qtypes.lua Normal file
View File

@@ -0,0 +1,16 @@
---@class rtu_unit_qtypes
local qtypes = {}
local TBV_RTU_S_CMDS = {
INC_DUMP_MODE = 1,
DEC_DUMP_MODE = 2
}
local TBV_RTU_S_DATA = {
SET_DUMP_MODE = 1
}
qtypes.TBV_RTU_S_CMDS = TBV_RTU_S_CMDS
qtypes.TBV_RTU_S_DATA = TBV_RTU_S_DATA
return qtypes

330
supervisor/session/rtu/redstone.lua
View File

@@ -12,39 +12,40 @@ local redstone = {}
local RTU_UNIT_TYPES = comms.RTU_UNIT_TYPES
local MODBUS_FCODE = types.MODBUS_FCODE
local RS_IO = rsio.IO
local IO_PORT = rsio.IO
local IO_LVL = rsio.IO_LVL
local IO_DIR = rsio.IO_DIR
local IO_MODE = rsio.IO_MODE
local RS_RTU_S_CMDS = {
}
local RS_RTU_S_DATA = {
RS_COMMAND = 1
}
redstone.RS_RTU_S_CMDS = RS_RTU_S_CMDS
redstone.RS_RTU_S_DATA = RS_RTU_S_DATA
local TXN_READY = -1
local TXN_TYPES = {
DI_READ = 1,
COIL_WRITE = 2,
INPUT_REG_READ = 3,
HOLD_REG_WRITE = 4
COIL_READ = 3,
INPUT_REG_READ = 4,
HOLD_REG_WRITE = 5,
HOLD_REG_READ = 6
}
local TXN_TAGS = {
"redstone.di_read",
"redstone.coil_write",
"redstone.input_reg_write",
"redstone.hold_reg_write"
"redstone.coil_read",
"redstone.input_reg_read",
"redstone.hold_reg_write",
"redstone.hold_reg_read"
}
local PERIODICS = {
INPUT_READ = 200
INPUT_READ = 200,
OUTPUT_SYNC = 200
}
---@class phy_entry
---@field phy IO_LVL
---@field req IO_LVL
-- create a new redstone rtu session runner
---@param session_id integer
---@param unit_id integer
@@ -62,57 +63,127 @@ function redstone.new(session_id, unit_id, advert, out_queue)
local self = {
session = unit_session.new(session_id, unit_id, advert, out_queue, log_tag, TXN_TAGS),
in_q = mqueue.new(),
has_di = false,
has_do = false,
has_ai = false,
has_ao = false,
periodics = {
next_di_req = 0,
next_ir_req = 0
next_di_req = 0,
next_cl_sync = 0,
next_ir_req = 0,
next_hr_sync = 0
},
---@class rs_io_list
io_list = {
digital_in = {}, -- discrete inputs
digital_out = {}, -- coils
analog_in = {}, -- input registers
analog_out = {} -- holding registers
digital_in = {}, -- discrete inputs
digital_out = {}, -- coils
analog_in = {}, -- input registers
analog_out = {} -- holding registers
},
db = {}
phy_trans = { coils = -1, hold_regs = -1 },
-- last set/read ports (reflecting the current state of the RTU)
---@class rs_io_states
phy_io = {
digital_in = {}, -- discrete inputs
digital_out = {}, -- coils
analog_in = {}, -- input registers
analog_out = {} -- holding registers
},
---@class redstone_session_db
db = {
-- read/write functions for connected I/O
io = {}
}
}
local public = self.session.get()
-- INITIALIZE --
-- create all channels as disconnected
for _ = 1, #RS_IO do
-- create all ports as disconnected
for _ = 1, #IO_PORT do
table.insert(self.db, IO_LVL.DISCONNECT)
end
-- setup I/O
for i = 1, #advert.rsio do
local channel = advert.rsio[i]
local port = advert.rsio[i]
if rsio.is_valid_channel(channel) then
local mode = rsio.get_io_mode(channel)
if rsio.is_valid_port(port) then
local mode = rsio.get_io_mode(port)
if mode == IO_MODE.DIGITAL_IN then
self.has_di = true
table.insert(self.io_list.digital_in, channel)
table.insert(self.io_list.digital_in, port)
self.phy_io.digital_in[port] = { phy = IO_LVL.FLOATING, req = IO_LVL.FLOATING }
---@class rs_db_dig_io
local io_f = {
read = function () return rsio.digital_is_active(port, self.phy_io.digital_in[port].phy) end,
---@param active boolean
write = function (active) end
}
self.db.io[port] = io_f
elseif mode == IO_MODE.DIGITAL_OUT then
table.insert(self.io_list.digital_out, channel)
self.has_do = true
table.insert(self.io_list.digital_out, port)
self.phy_io.digital_out[port] = { phy = IO_LVL.FLOATING, req = IO_LVL.FLOATING }
---@class rs_db_dig_io
local io_f = {
read = function () return rsio.digital_is_active(port, self.phy_io.digital_out[port].phy) end,
---@param active boolean
write = function (active)
local level = rsio.digital_write_active(port, active)
if level ~= nil then self.phy_io.digital_out[port].req = level end
end
}
self.db.io[port] = io_f
elseif mode == IO_MODE.ANALOG_IN then
self.has_ai = true
table.insert(self.io_list.analog_in, channel)
table.insert(self.io_list.analog_in, port)
self.phy_io.analog_in[port] = { phy = 0, req = 0 }
---@class rs_db_ana_io
local io_f = {
---@return integer
read = function () return self.phy_io.analog_in[port].phy end,
---@param value integer
write = function (value) end
}
self.db.io[port] = io_f
elseif mode == IO_MODE.ANALOG_OUT then
table.insert(self.io_list.analog_out, channel)
self.has_ao = true
table.insert(self.io_list.analog_out, port)
self.phy_io.analog_out[port] = { phy = 0, req = 0 }
---@class rs_db_ana_io
local io_f = {
---@return integer
read = function () return self.phy_io.analog_out[port].phy end,
---@param value integer
write = function (value)
if value >= 0 and value <= 15 then
self.phy_io.analog_out[port].req = value
end
end
}
self.db.io[port] = io_f
else
-- should be unreachable code, we already validated channels
log.error(util.c(log_tag, "failed to identify advertisement channel IO mode (", channel, ")"), true)
-- should be unreachable code, we already validated ports
log.error(util.c(log_tag, "failed to identify advertisement port IO mode (", port, ")"), true)
return nil
end
self.db[channel] = IO_LVL.LOW
else
log.error(util.c(log_tag, "invalid advertisement channel (", channel, ")"), true)
log.error(util.c(log_tag, "invalid advertisement port (", port, ")"), true)
return nil
end
end
@@ -129,14 +200,40 @@ function redstone.new(session_id, unit_id, advert, out_queue)
self.session.send_request(TXN_TYPES.INPUT_REG_READ, MODBUS_FCODE.READ_INPUT_REGS, { 1, #self.io_list.analog_in })
end
-- write coil output
local function _write_coil(coil, value)
self.session.send_request(TXN_TYPES.COIL_WRITE, MODBUS_FCODE.WRITE_MUL_COILS, { coil, value })
-- write all coil outputs
local function _write_coils()
local params = { 1 }
local outputs = self.phy_io.digital_out
for i = 1, #self.io_list.digital_out do
local port = self.io_list.digital_out[i]
table.insert(params, outputs[port].req)
end
self.phy_trans.coils = self.session.send_request(TXN_TYPES.COIL_WRITE, MODBUS_FCODE.WRITE_MUL_COILS, params)
end
-- write holding register output
local function _write_holding_register(reg, value)
self.session.send_request(TXN_TYPES.HOLD_REG_WRITE, MODBUS_FCODE.WRITE_MUL_HOLD_REGS, { reg, value })
-- read all coil outputs
local function _read_coils()
self.session.send_request(TXN_TYPES.COIL_READ, MODBUS_FCODE.READ_COILS, { 1, #self.io_list.digital_out })
end
-- write all holding register outputs
local function _write_holding_registers()
local params = { 1 }
local outputs = self.phy_io.analog_out
for i = 1, #self.io_list.analog_out do
local port = self.io_list.analog_out[i]
table.insert(params, outputs[port].req)
end
self.phy_trans.hold_regs = self.session.send_request(TXN_TYPES.HOLD_REG_WRITE, MODBUS_FCODE.WRITE_MUL_HOLD_REGS, params)
end
-- read all holding register outputs
local function _read_holding_registers()
self.session.send_request(TXN_TYPES.HOLD_REG_READ, MODBUS_FCODE.READ_MUL_HOLD_REGS, { 1, #self.io_list.analog_out })
end
-- PUBLIC FUNCTIONS --
@@ -146,14 +243,23 @@ function redstone.new(session_id, unit_id, advert, out_queue)
function public.handle_packet(m_pkt)
local txn_type = self.session.try_resolve(m_pkt)
if txn_type == false then
-- nothing to do
-- check if this is a failed write request
-- redstone operations are always immediately executed, so this would not be from an ACK or BUSY
if m_pkt.txn_id == self.phy_trans.coils then
self.phy_trans.coils = TXN_READY
log.debug(log_tag .. "failed to write coils, retrying soon")
elseif m_pkt.txn_id == self.phy_trans.hold_regs then
self.phy_trans.hold_regs = TXN_READY
log.debug(log_tag .. "failed to write holding registers, retrying soon")
end
elseif txn_type == TXN_TYPES.DI_READ then
-- discrete input read response
if m_pkt.length == #self.io_list.digital_in then
for i = 1, m_pkt.length do
local channel = self.io_list.digital_in[i]
local port = self.io_list.digital_in[i]
local value = m_pkt.data[i]
self.db[channel] = value
self.phy_io.digital_in[port].phy = value
end
else
log.debug(log_tag .. "MODBUS transaction reply length mismatch (" .. TXN_TAGS[txn_type] .. ")")
@@ -162,15 +268,55 @@ function redstone.new(session_id, unit_id, advert, out_queue)
-- input register read response
if m_pkt.length == #self.io_list.analog_in then
for i = 1, m_pkt.length do
local channel = self.io_list.analog_in[i]
local port = self.io_list.analog_in[i]
local value = m_pkt.data[i]
self.db[channel] = value
self.phy_io.analog_in[port].phy = value
end
else
log.debug(log_tag .. "MODBUS transaction reply length mismatch (" .. TXN_TAGS[txn_type] .. ")")
end
elseif txn_type == TXN_TYPES.COIL_WRITE or txn_type == TXN_TYPES.HOLD_REG_WRITE then
-- successful acknowledgement
elseif txn_type == TXN_TYPES.COIL_WRITE then
-- successful acknowledgement, read back
_read_coils()
elseif txn_type == TXN_TYPES.COIL_READ then
-- update phy I/O table
-- if there are multiple outputs for the same port, they will overwrite eachother (but *should* be identical)
-- given these are redstone outputs, if one worked they all should have, so no additional verification will be done
if m_pkt.length == #self.io_list.digital_out then
for i = 1, m_pkt.length do
local port = self.io_list.digital_out[i]
local value = m_pkt.data[i]
self.phy_io.digital_out[port].phy = value
if self.phy_io.digital_out[port].req == IO_LVL.FLOATING then
self.phy_io.digital_out[port].req = value
end
end
self.phy_trans.coils = TXN_READY
else
log.debug(log_tag .. "MODBUS transaction reply length mismatch (" .. TXN_TAGS[txn_type] .. ")")
end
elseif txn_type == TXN_TYPES.HOLD_REG_WRITE then
-- successful acknowledgement, read back
_read_holding_registers()
elseif txn_type == TXN_TYPES.HOLD_REG_READ then
-- update phy I/O table
-- if there are multiple outputs for the same port, they will overwrite eachother (but *should* be identical)
-- given these are redstone outputs, if one worked they all should have, so no additional verification will be done
if m_pkt.length == #self.io_list.analog_out then
for i = 1, m_pkt.length do
local port = self.io_list.analog_out[i]
local value = m_pkt.data[i]
self.phy_io.analog_out[port].phy = value
end
else
log.debug(log_tag .. "MODBUS transaction reply length mismatch (" .. TXN_TAGS[txn_type] .. ")")
end
self.phy_trans.hold_regs = TXN_READY
elseif txn_type == nil then
log.error(log_tag .. "unknown transaction reply")
else
@@ -181,60 +327,6 @@ function redstone.new(session_id, unit_id, advert, out_queue)
-- update this runner
---@param time_now integer milliseconds
function public.update(time_now)
-- check command queue
while self.in_q.ready() do
-- get a new message to process
local msg = self.in_q.pop()
if msg ~= nil then
if msg.qtype == mqueue.TYPE.DATA then
-- instruction with body
local cmd = msg.message ---@type queue_data
if cmd.key == RS_RTU_S_DATA.RS_COMMAND then
local rs_cmd = cmd.val ---@type rs_session_command
if self.db[rs_cmd.channel] ~= IO_LVL.DISCONNECT then
-- we have this as a connected channel
local mode = rsio.get_io_mode(rs_cmd.channel)
if mode == IO_MODE.DIGITAL_OUT then
-- record the value for retries
self.db[rs_cmd.channel] = rs_cmd.value
-- find the coil address then write to it
for i = 0, #self.digital_out do
if self.digital_out[i] == rs_cmd.channel then
_write_coil(i, rs_cmd.value)
break
end
end
elseif mode == IO_MODE.ANALOG_OUT then
-- record the value for retries
self.db[rs_cmd.channel] = rs_cmd.value
-- find the holding register address then write to it
for i = 0, #self.analog_out do
if self.analog_out[i] == rs_cmd.channel then
_write_holding_register(i, rs_cmd.value)
break
end
end
else
log.debug(util.c(log_tag, "attemted write to non D/O or A/O mode ", mode))
end
end
end
end
end
-- max 100ms spent processing queue
if util.time() - time_now > 100 then
log.warning(log_tag .. "exceeded 100ms queue process limit")
break
end
end
time_now = util.time()
-- poll digital inputs
if self.has_di then
if self.periodics.next_di_req <= time_now then
@@ -243,6 +335,20 @@ function redstone.new(session_id, unit_id, advert, out_queue)
end
end
-- sync digital outputs
if self.has_do then
if (self.periodics.next_cl_sync <= time_now) and (self.phy_trans.coils == TXN_READY) then
for _, entry in pairs(self.phy_io.digital_out) do
if entry.phy ~= entry.req then
_write_coils()
break
end
end
self.periodics.next_cl_sync = time_now + PERIODICS.OUTPUT_SYNC
end
end
-- poll analog inputs
if self.has_ai then
if self.periodics.next_ir_req <= time_now then
@@ -251,6 +357,20 @@ function redstone.new(session_id, unit_id, advert, out_queue)
end
end
-- sync analog outputs
if self.has_ao then
if (self.periodics.next_hr_sync <= time_now) and (self.phy_trans.hold_regs == TXN_READY) then
for _, entry in pairs(self.phy_io.analog_out) do
if entry.phy ~= entry.req then
_write_holding_registers()
break
end
end
self.periodics.next_hr_sync = time_now + PERIODICS.OUTPUT_SYNC
end
end
self.session.post_update()
end
@@ -262,7 +382,7 @@ function redstone.new(session_id, unit_id, advert, out_queue)
-- get the unit session database
function public.get_db() return self.db end
return public, self.in_q
return public
end
return redstone

33
supervisor/session/rtu/turbinev.lua
View File

@@ -4,6 +4,7 @@ local mqueue = require("scada-common.mqueue")
local types = require("scada-common.types")
local util = require("scada-common.util")
local qtypes = require("supervisor.session.rtu.qtypes")
local unit_session = require("supervisor.session.rtu.unit_session")
local turbinev = {}
@@ -12,17 +13,8 @@ local RTU_UNIT_TYPES = comms.RTU_UNIT_TYPES
local DUMPING_MODE = types.DUMPING_MODE
local MODBUS_FCODE = types.MODBUS_FCODE
local TBV_RTU_S_CMDS = {
INC_DUMP_MODE = 1,
DEC_DUMP_MODE = 2
}
local TBV_RTU_S_DATA = {
SET_DUMP_MODE = 1
}
turbinev.RS_RTU_S_CMDS = TBV_RTU_S_CMDS
turbinev.RS_RTU_S_DATA = TBV_RTU_S_DATA
local TBV_RTU_S_CMDS = qtypes.TBV_RTU_S_CMDS
local TBV_RTU_S_DATA = qtypes.TBV_RTU_S_DATA
local TXN_TYPES = {
FORMED = 1,
@@ -67,7 +59,6 @@ function turbinev.new(session_id, unit_id, advert, out_queue)
local self = {
session = unit_session.new(session_id, unit_id, advert, out_queue, log_tag, TXN_TAGS),
in_q = mqueue.new(),
has_build = false,
periodics = {
next_formed_req = 0,
@@ -240,9 +231,9 @@ function turbinev.new(session_id, unit_id, advert, out_queue)
---@param time_now integer milliseconds
function public.update(time_now)
-- check command queue
while self.in_q.ready() do
while self.session.in_q.ready() do
-- get a new message to process
local msg = self.in_q.pop()
local msg = self.session.in_q.pop()
if msg ~= nil then
if msg.qtype == mqueue.TYPE.COMMAND then
@@ -254,15 +245,21 @@ function turbinev.new(session_id, unit_id, advert, out_queue)
elseif cmd == TBV_RTU_S_CMDS.DEC_DUMP_MODE then
_dec_dump_mode()
else
log.debug(util.c(log_tag, "unrecognized in_q command ", cmd))
log.debug(util.c(log_tag, "unrecognized in-queue command ", cmd))
end
elseif msg.qtype == mqueue.TYPE.DATA then
-- instruction with body
local cmd = msg.message ---@type queue_data
if cmd.key == TBV_RTU_S_DATA.SET_DUMP_MODE then
_set_dump_mode(cmd.val)
if cmd.val == types.DUMPING_MODE.IDLE or
cmd.val == types.DUMPING_MODE.DUMPING_EXCESS or
cmd.val == types.DUMPING_MODE.DUMPING then
_set_dump_mode(cmd.val)
else
log.debug(util.c(log_tag, "unrecognized dumping mode \"", cmd.val, "\""))
end
else
log.debug(util.c(log_tag, "unrecognized in_q data ", cmd.key))
log.debug(util.c(log_tag, "unrecognized in-queue data ", cmd.key))
end
end
end
@@ -313,7 +310,7 @@ function turbinev.new(session_id, unit_id, advert, out_queue)
-- get the unit session database
function public.get_db() return self.db end
return public, self.in_q
return public
end
return turbinev

21
supervisor/session/rtu/unit_session.lua
View File

@@ -1,5 +1,6 @@
local comms = require("scada-common.comms")
local log = require("scada-common.log")
local mqueue = require("scada-common.mqueue")
local types = require("scada-common.types")
local util = require("scada-common.util")
@@ -42,7 +43,9 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
}
---@class _unit_session
local protected = {}
local protected = {
in_q = mqueue.new()
}
---@class unit_session
local public = {}
@@ -53,6 +56,7 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
---@param txn_type integer transaction type
---@param f_code MODBUS_FCODE function code
---@param register_param table register range or register and values
---@return integer txn_id transaction ID of this transaction
function protected.send_request(txn_type, f_code, register_param)
local m_pkt = comms.modbus_packet()
local txn_id = self.transaction_controller.create(txn_type)
@@ -60,11 +64,13 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
m_pkt.make(txn_id, self.unit_id, f_code, register_param)
self.out_q.push_packet(m_pkt)
return txn_id
end
-- try to resolve a MODBUS transaction
---@param m_pkt modbus_frame MODBUS packet
---@return integer|false txn_type transaction type or false on error/busy
---@return integer|false txn_type, integer txn_id transaction type or false on error/busy, transaction ID
function protected.try_resolve(m_pkt)
if m_pkt.scada_frame.protocol() == PROTOCOLS.MODBUS_TCP then
if m_pkt.unit_id == self.unit_id then
@@ -110,7 +116,7 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
self.device_fail = false
-- no error, return the transaction type
return txn_type
return txn_type, m_pkt.txn_id
end
else
log.error(log_tag .. "wrong unit ID: " .. m_pkt.unit_id, true)
@@ -120,7 +126,7 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
end
-- error or transaction in progress, return false
return false
return false, m_pkt.txn_id
end
-- post update tasks
@@ -141,6 +147,8 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
function public.get_device_idx() return self.device_index end
-- get the reactor ID
function public.get_reactor() return self.reactor end
-- get the command queue
function public.get_cmd_queue() return protected.in_q end
-- close this unit
function public.close() self.connected = false end
@@ -171,7 +179,10 @@ function unit_session.new(session_id, unit_id, advert, out_queue, log_tag, txn_t
end
-- get the unit session database
function public.get_db() return {} end
function public.get_db()
log.debug("template unit_session.get_db() called", true)
return {}
end
return protected
end