mirror of
https://github.com/MikaylaFischler/cc-mek-scada.git
synced 2024-08-30 18:22:34 +00:00
543 lines
17 KiB
Lua
543 lines
17 KiB
Lua
local log = require("scada-common.log")
|
|
local mqueue = require("scada-common.mqueue")
|
|
local util = require("scada-common.util")
|
|
|
|
local databus = require("supervisor.databus")
|
|
local facility = require("supervisor.facility")
|
|
local pgi = require("supervisor.pgi")
|
|
|
|
local coordinator = require("supervisor.session.coordinator")
|
|
local plc = require("supervisor.session.plc")
|
|
local pocket = require("supervisor.session.pocket")
|
|
local rtu = require("supervisor.session.rtu")
|
|
local svqtypes = require("supervisor.session.svqtypes")
|
|
|
|
-- Supervisor Sessions Handler
|
|
|
|
local SV_Q_DATA = svqtypes.SV_Q_DATA
|
|
|
|
local PLC_S_CMDS = plc.PLC_S_CMDS
|
|
local PLC_S_DATA = plc.PLC_S_DATA
|
|
local CRD_S_DATA = coordinator.CRD_S_DATA
|
|
|
|
local svsessions = {}
|
|
|
|
---@enum SESSION_TYPE
|
|
local SESSION_TYPE = {
|
|
RTU_SESSION = 0, -- RTU gateway
|
|
PLC_SESSION = 1, -- reactor PLC
|
|
CRD_SESSION = 2, -- coordinator
|
|
PDG_SESSION = 3 -- pocket diagnostics
|
|
}
|
|
|
|
svsessions.SESSION_TYPE = SESSION_TYPE
|
|
|
|
local self = {
|
|
nic = nil, ---@type nic|nil
|
|
fp_ok = false,
|
|
config = nil, ---@type svr_config
|
|
facility = nil, ---@type facility|nil
|
|
sessions = { rtu = {}, plc = {}, crd = {}, pdg = {} },
|
|
next_ids = { rtu = 0, plc = 0, crd = 0, pdg = 0 },
|
|
dev_dbg = { duplicate = {}, out_of_range = {} }
|
|
}
|
|
|
|
---@alias sv_session_structs plc_session_struct|rtu_session_struct|crd_session_struct|pdg_session_struct
|
|
|
|
-- PRIVATE FUNCTIONS --
|
|
|
|
-- handle a session output queue
|
|
---@param session sv_session_structs
|
|
local function _sv_handle_outq(session)
|
|
-- record handler start time
|
|
local handle_start = util.time()
|
|
|
|
-- process output queue
|
|
while session.out_queue.ready() do
|
|
-- get a new message to process
|
|
local msg = session.out_queue.pop()
|
|
|
|
if msg ~= nil then
|
|
if msg.qtype == mqueue.TYPE.PACKET then
|
|
-- handle a packet to be sent
|
|
self.nic.transmit(session.r_chan, self.config.SVR_Channel, msg.message)
|
|
elseif msg.qtype == mqueue.TYPE.COMMAND then
|
|
-- handle instruction/notification
|
|
elseif msg.qtype == mqueue.TYPE.DATA then
|
|
-- instruction/notification with body
|
|
local cmd = msg.message ---@type queue_data
|
|
|
|
if cmd.key < SV_Q_DATA.__END_PLC_CMDS__ then
|
|
-- PLC commands from coordinator
|
|
local plc_s = svsessions.get_reactor_session(cmd.val[1])
|
|
|
|
if plc_s ~= nil then
|
|
if cmd.key == SV_Q_DATA.START then
|
|
plc_s.in_queue.push_command(PLC_S_CMDS.ENABLE)
|
|
elseif cmd.key == SV_Q_DATA.SCRAM then
|
|
plc_s.in_queue.push_command(PLC_S_CMDS.SCRAM)
|
|
elseif cmd.key == SV_Q_DATA.RESET_RPS then
|
|
plc_s.in_queue.push_command(PLC_S_CMDS.RPS_RESET)
|
|
elseif cmd.key == SV_Q_DATA.SET_BURN and type(cmd.val) == "table" and #cmd.val == 2 then
|
|
plc_s.in_queue.push_data(PLC_S_DATA.BURN_RATE, cmd.val[2])
|
|
else
|
|
log.debug(util.c("SVS: unknown PLC SV queue command ", cmd.key))
|
|
end
|
|
end
|
|
else
|
|
local crd_s = svsessions.get_crd_session()
|
|
if crd_s ~= nil then
|
|
if cmd.key == SV_Q_DATA.CRDN_ACK then
|
|
-- ack to be sent to coordinator
|
|
crd_s.in_queue.push_data(CRD_S_DATA.CMD_ACK, cmd.val)
|
|
elseif cmd.key == SV_Q_DATA.PLC_BUILD_CHANGED then
|
|
-- a PLC build has changed
|
|
crd_s.in_queue.push_data(CRD_S_DATA.RESEND_PLC_BUILD, cmd.val)
|
|
elseif cmd.key == SV_Q_DATA.RTU_BUILD_CHANGED then
|
|
-- an RTU build has changed
|
|
crd_s.in_queue.push_data(CRD_S_DATA.RESEND_RTU_BUILD, cmd.val)
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
-- max 100ms spent processing queue
|
|
if util.time() - handle_start > 100 then
|
|
log.debug("SVS: supervisor out queue handler exceeded 100ms queue process limit")
|
|
log.debug(util.c("SVS: offending session: ", session))
|
|
break
|
|
end
|
|
end
|
|
end
|
|
|
|
-- iterate all the given sessions
|
|
---@param sessions table
|
|
local function _iterate(sessions)
|
|
for i = 1, #sessions do
|
|
local session = sessions[i] ---@type sv_session_structs
|
|
|
|
if session.open and session.instance.iterate() then
|
|
_sv_handle_outq(session)
|
|
else
|
|
session.open = false
|
|
end
|
|
end
|
|
end
|
|
|
|
-- cleanly close a session
|
|
---@param session sv_session_structs
|
|
local function _shutdown(session)
|
|
session.open = false
|
|
session.instance.close()
|
|
|
|
-- send packets in out queue (for the close packet)
|
|
while session.out_queue.ready() do
|
|
local msg = session.out_queue.pop()
|
|
if msg ~= nil and msg.qtype == mqueue.TYPE.PACKET then
|
|
self.nic.transmit(session.r_chan, self.config.SVR_Channel, msg.message)
|
|
end
|
|
end
|
|
|
|
log.debug(util.c("SVS: closed session ", session))
|
|
end
|
|
|
|
-- close connections
|
|
---@param sessions table
|
|
local function _close(sessions)
|
|
for i = 1, #sessions do
|
|
local session = sessions[i] ---@type sv_session_structs
|
|
if session.open then _shutdown(session) end
|
|
end
|
|
end
|
|
|
|
-- check if a watchdog timer event matches that of one of the provided sessions
|
|
---@param sessions table
|
|
---@param timer_event number
|
|
local function _check_watchdogs(sessions, timer_event)
|
|
for i = 1, #sessions do
|
|
local session = sessions[i] ---@type sv_session_structs
|
|
if session.open then
|
|
local triggered = session.instance.check_wd(timer_event)
|
|
if triggered then
|
|
log.debug(util.c("SVS: watchdog closing session ", session, "..."))
|
|
_shutdown(session)
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
-- delete any closed sessions
|
|
---@param sessions table
|
|
local function _free_closed(sessions)
|
|
local f = function (session) return session.open end
|
|
|
|
---@param session sv_session_structs
|
|
local on_delete = function (session)
|
|
log.debug(util.c("SVS: free'ing closed session ", session))
|
|
end
|
|
|
|
util.filter_table(sessions, f, on_delete)
|
|
end
|
|
|
|
-- find a session by computer ID
|
|
---@nodiscard
|
|
---@param list table
|
|
---@param s_addr integer
|
|
---@return sv_session_structs|nil
|
|
local function _find_session(list, s_addr)
|
|
for i = 1, #list do
|
|
if list[i].s_addr == s_addr then return list[i] end
|
|
end
|
|
return nil
|
|
end
|
|
|
|
local function _update_dev_dbg()
|
|
local f = function (unit) return unit.is_connected() end
|
|
|
|
util.filter_table(self.dev_dbg.duplicate, f, pgi.delete_chk_entry)
|
|
util.filter_table(self.dev_dbg.out_of_range, f, pgi.delete_chk_entry)
|
|
end
|
|
|
|
-- SHARED FUNCTIONS --
|
|
|
|
---@param unit unit_session RTU session
|
|
---@param list table table of RTU sessions
|
|
---@param max integer max of this type of RTU
|
|
---@return 0|1|2|3 fail_code, string fail_str 0 = success, 1 = out-of-range, 2 = duplicate, 3 = exceeded table max
|
|
local function check_rtu_id(unit, list, max)
|
|
local fail_code, fail_str = 0, "OK"
|
|
|
|
if (unit.get_device_idx() < 1 and max ~= 1) or unit.get_device_idx() > max then
|
|
-- out-of-range
|
|
fail_code, fail_str = 1, "index out of range"
|
|
table.insert(self.dev_dbg.out_of_range, unit)
|
|
else
|
|
for _, u in ipairs(list) do
|
|
if u.get_device_idx() == unit.get_device_idx() then
|
|
-- duplicate
|
|
fail_code, fail_str = 2, "duplicate index"
|
|
table.insert(self.dev_dbg.duplicate, unit)
|
|
break
|
|
end
|
|
end
|
|
end
|
|
|
|
-- make sure this won't exceed the maximum allowable devices
|
|
if fail_code == 0 and #list >= max then
|
|
fail_code, fail_str = 3, "too many of this type"
|
|
end
|
|
|
|
-- add to the list for the user
|
|
if fail_code > 0 then
|
|
local cmp_id
|
|
|
|
for i = 1, #self.sessions.rtu do
|
|
if self.sessions.rtu[i].instance.get_id() == unit.get_session_id() then
|
|
cmp_id = self.sessions.rtu[i].s_addr
|
|
break
|
|
end
|
|
end
|
|
|
|
pgi.create_chk_entry(unit, fail_code, cmp_id)
|
|
end
|
|
|
|
return fail_code, fail_str
|
|
end
|
|
|
|
-- PUBLIC FUNCTIONS --
|
|
|
|
-- initialize svsessions
|
|
---@param nic nic network interface device
|
|
---@param fp_ok boolean front panel active
|
|
---@param config svr_config supervisor configuration
|
|
---@param cooling_conf sv_cooling_conf cooling configuration definition
|
|
function svsessions.init(nic, fp_ok, config, cooling_conf)
|
|
self.nic = nic
|
|
self.fp_ok = fp_ok
|
|
self.config = config
|
|
self.facility = facility.new(config, cooling_conf, check_rtu_id)
|
|
end
|
|
|
|
-- find an RTU session by the computer ID
|
|
---@nodiscard
|
|
---@param source_addr integer
|
|
---@return rtu_session_struct|nil
|
|
function svsessions.find_rtu_session(source_addr)
|
|
-- check RTU sessions
|
|
local session = _find_session(self.sessions.rtu, source_addr)
|
|
---@cast session rtu_session_struct|nil
|
|
return session
|
|
end
|
|
|
|
-- find a PLC session by the computer ID
|
|
---@nodiscard
|
|
---@param source_addr integer
|
|
---@return plc_session_struct|nil
|
|
function svsessions.find_plc_session(source_addr)
|
|
-- check PLC sessions
|
|
local session = _find_session(self.sessions.plc, source_addr)
|
|
---@cast session plc_session_struct|nil
|
|
return session
|
|
end
|
|
|
|
-- find a coordinator session by the computer ID
|
|
---@nodiscard
|
|
---@param source_addr integer
|
|
---@return crd_session_struct|nil
|
|
function svsessions.find_crd_session(source_addr)
|
|
-- check coordinator sessions
|
|
local session = _find_session(self.sessions.crd, source_addr)
|
|
---@cast session crd_session_struct|nil
|
|
return session
|
|
end
|
|
|
|
-- find a pocket diagnostics session by the computer ID
|
|
---@nodiscard
|
|
---@param source_addr integer
|
|
---@return pdg_session_struct|nil
|
|
function svsessions.find_pdg_session(source_addr)
|
|
-- check diagnostic sessions
|
|
local session = _find_session(self.sessions.pdg, source_addr)
|
|
---@cast session pdg_session_struct|nil
|
|
return session
|
|
end
|
|
|
|
-- get the a coordinator session if exists
|
|
---@nodiscard
|
|
---@return crd_session_struct|nil
|
|
function svsessions.get_crd_session()
|
|
return self.sessions.crd[1]
|
|
end
|
|
|
|
-- get a session by reactor ID
|
|
---@nodiscard
|
|
---@param reactor integer
|
|
---@return plc_session_struct|nil session
|
|
function svsessions.get_reactor_session(reactor)
|
|
local session = nil
|
|
|
|
for i = 1, #self.sessions.plc do
|
|
if self.sessions.plc[i].reactor == reactor then
|
|
session = self.sessions.plc[i]
|
|
end
|
|
end
|
|
|
|
return session
|
|
end
|
|
|
|
-- establish a new PLC session
|
|
---@nodiscard
|
|
---@param source_addr integer PLC computer ID
|
|
---@param i_seq_num integer initial (most recent) sequence number
|
|
---@param for_reactor integer unit ID
|
|
---@param version string PLC version
|
|
---@return integer|false session_id
|
|
function svsessions.establish_plc_session(source_addr, i_seq_num, for_reactor, version)
|
|
if svsessions.get_reactor_session(for_reactor) == nil and for_reactor >= 1 and for_reactor <= self.config.UnitCount then
|
|
---@class plc_session_struct
|
|
local plc_s = {
|
|
s_type = "plc",
|
|
open = true,
|
|
reactor = for_reactor,
|
|
version = version,
|
|
r_chan = self.config.PLC_Channel,
|
|
s_addr = source_addr,
|
|
in_queue = mqueue.new(),
|
|
out_queue = mqueue.new(),
|
|
instance = nil ---@type plc_session
|
|
}
|
|
|
|
local id = self.next_ids.plc
|
|
|
|
plc_s.instance = plc.new_session(id, source_addr, i_seq_num, for_reactor, plc_s.in_queue, plc_s.out_queue, self.config.PLC_Timeout, self.fp_ok)
|
|
table.insert(self.sessions.plc, plc_s)
|
|
|
|
local units = self.facility.get_units()
|
|
units[for_reactor].link_plc_session(plc_s)
|
|
|
|
local mt = {
|
|
---@param s plc_session_struct
|
|
__tostring = function (s) return util.c("PLC [", s.instance.get_id(), "] for reactor #", s.reactor, " (@", s.s_addr, ")") end
|
|
}
|
|
|
|
setmetatable(plc_s, mt)
|
|
|
|
databus.tx_plc_connected(for_reactor, version, source_addr)
|
|
log.debug(util.c("SVS: established new session: ", plc_s))
|
|
|
|
self.next_ids.plc = id + 1
|
|
|
|
-- success
|
|
return plc_s.instance.get_id()
|
|
else
|
|
-- reactor already assigned to a PLC or ID out of range
|
|
return false
|
|
end
|
|
end
|
|
|
|
-- establish a new RTU gateway session
|
|
---@nodiscard
|
|
---@param source_addr integer RTU gateway computer ID
|
|
---@param i_seq_num integer initial (most recent) sequence number
|
|
---@param advertisement table RTU capability advertisement
|
|
---@param version string RTU gateway version
|
|
---@return integer session_id
|
|
function svsessions.establish_rtu_session(source_addr, i_seq_num, advertisement, version)
|
|
---@class rtu_session_struct
|
|
local rtu_s = {
|
|
s_type = "rtu",
|
|
open = true,
|
|
version = version,
|
|
r_chan = self.config.RTU_Channel,
|
|
s_addr = source_addr,
|
|
in_queue = mqueue.new(),
|
|
out_queue = mqueue.new(),
|
|
instance = nil ---@type rtu_session
|
|
}
|
|
|
|
local id = self.next_ids.rtu
|
|
|
|
rtu_s.instance = rtu.new_session(id, source_addr, i_seq_num, rtu_s.in_queue, rtu_s.out_queue, self.config.RTU_Timeout, advertisement, self.facility, self.fp_ok)
|
|
table.insert(self.sessions.rtu, rtu_s)
|
|
|
|
local mt = {
|
|
---@param s rtu_session_struct
|
|
__tostring = function (s) return util.c("RTU [", s.instance.get_id(), "] (@", s.s_addr, ")") end
|
|
}
|
|
|
|
setmetatable(rtu_s, mt)
|
|
|
|
databus.tx_rtu_connected(id, version, source_addr)
|
|
log.debug(util.c("SVS: established new session: ", rtu_s))
|
|
|
|
self.next_ids.rtu = id + 1
|
|
|
|
-- success
|
|
return id
|
|
end
|
|
|
|
-- establish a new coordinator session
|
|
---@nodiscard
|
|
---@param source_addr integer coordinator computer ID
|
|
---@param i_seq_num integer initial (most recent) sequence number
|
|
---@param version string coordinator version
|
|
---@return integer|false session_id
|
|
function svsessions.establish_crd_session(source_addr, i_seq_num, version)
|
|
if svsessions.get_crd_session() == nil then
|
|
---@class crd_session_struct
|
|
local crd_s = {
|
|
s_type = "crd",
|
|
open = true,
|
|
version = version,
|
|
r_chan = self.config.CRD_Channel,
|
|
s_addr = source_addr,
|
|
in_queue = mqueue.new(),
|
|
out_queue = mqueue.new(),
|
|
instance = nil ---@type crd_session
|
|
}
|
|
|
|
local id = self.next_ids.crd
|
|
|
|
crd_s.instance = coordinator.new_session(id, source_addr, i_seq_num, crd_s.in_queue, crd_s.out_queue, self.config.CRD_Timeout, self.facility, self.fp_ok)
|
|
table.insert(self.sessions.crd, crd_s)
|
|
|
|
local mt = {
|
|
---@param s crd_session_struct
|
|
__tostring = function (s) return util.c("CRD [", s.instance.get_id(), "] (@", s.s_addr, ")") end
|
|
}
|
|
|
|
setmetatable(crd_s, mt)
|
|
|
|
databus.tx_crd_connected(version, source_addr)
|
|
log.debug(util.c("SVS: established new session: ", crd_s))
|
|
|
|
self.next_ids.crd = id + 1
|
|
|
|
-- success
|
|
return id
|
|
else
|
|
-- we already have a coordinator linked
|
|
return false
|
|
end
|
|
end
|
|
|
|
-- establish a new pocket diagnostics session
|
|
---@nodiscard
|
|
---@param source_addr integer pocket computer ID
|
|
---@param i_seq_num integer initial (most recent) sequence number
|
|
---@param version string pocket version
|
|
---@return integer|false session_id
|
|
function svsessions.establish_pdg_session(source_addr, i_seq_num, version)
|
|
---@class pdg_session_struct
|
|
local pdg_s = {
|
|
s_type = "pkt",
|
|
open = true,
|
|
version = version,
|
|
r_chan = self.config.PKT_Channel,
|
|
s_addr = source_addr,
|
|
in_queue = mqueue.new(),
|
|
out_queue = mqueue.new(),
|
|
instance = nil ---@type pdg_session
|
|
}
|
|
|
|
local id = self.next_ids.pdg
|
|
|
|
pdg_s.instance = pocket.new_session(id, source_addr, i_seq_num, pdg_s.in_queue, pdg_s.out_queue, self.config.PKT_Timeout, self.facility, self.fp_ok)
|
|
table.insert(self.sessions.pdg, pdg_s)
|
|
|
|
local mt = {
|
|
---@param s pdg_session_struct
|
|
__tostring = function (s) return util.c("PDG [", s.instance.get_id(), "] (@", s.s_addr, ")") end
|
|
}
|
|
|
|
setmetatable(pdg_s, mt)
|
|
|
|
databus.tx_pdg_connected(id, version, source_addr)
|
|
log.debug(util.c("SVS: established new session: ", pdg_s))
|
|
|
|
self.next_ids.pdg = id + 1
|
|
|
|
-- success
|
|
return id
|
|
end
|
|
|
|
-- attempt to identify which session's watchdog timer fired
|
|
---@param timer_event number
|
|
function svsessions.check_all_watchdogs(timer_event)
|
|
for _, list in pairs(self.sessions) do _check_watchdogs(list, timer_event) end
|
|
end
|
|
|
|
-- iterate all sessions, and update facility/unit data & process control logic
|
|
function svsessions.iterate_all()
|
|
-- iterate sessions
|
|
for _, list in pairs(self.sessions) do _iterate(list) end
|
|
|
|
-- report RTU sessions to facility
|
|
self.facility.report_rtus(self.sessions.rtu)
|
|
|
|
-- iterate facility
|
|
self.facility.update()
|
|
|
|
-- iterate units
|
|
self.facility.update_units()
|
|
|
|
_update_dev_dbg()
|
|
end
|
|
|
|
-- delete all closed sessions
|
|
function svsessions.free_all_closed()
|
|
for _, list in pairs(self.sessions) do _free_closed(list) end
|
|
end
|
|
|
|
-- close all open connections
|
|
function svsessions.close_all()
|
|
-- close sessions
|
|
for _, list in pairs(self.sessions) do _close(list) end
|
|
|
|
-- free sessions
|
|
svsessions.free_all_closed()
|
|
end
|
|
|
|
return svsessions
|