local audio = require("scada-common.audio") local comms = require("scada-common.comms") local ppm = require("scada-common.ppm") local log = require("scada-common.log") local types = require("scada-common.types") local util = require("scada-common.util") local themes = require("graphics.themes") local databus = require("rtu.databus") local modbus = require("rtu.modbus") local rtu = {} local PROTOCOL = comms.PROTOCOL local DEVICE_TYPE = comms.DEVICE_TYPE local ESTABLISH_ACK = comms.ESTABLISH_ACK local MGMT_TYPE = comms.MGMT_TYPE local RTU_UNIT_TYPE = types.RTU_UNIT_TYPE ---@type rtu_config local config = {} rtu.config = config -- load the RTU configuration function rtu.load_config() if not settings.load("/rtu.settings") then return false end config.Peripherals = settings.get("Peripherals") config.Redstone = settings.get("Redstone") config.SpeakerVolume = settings.get("SpeakerVolume") config.SVR_Channel = settings.get("SVR_Channel") config.RTU_Channel = settings.get("RTU_Channel") config.ConnTimeout = settings.get("ConnTimeout") config.TrustedRange = settings.get("TrustedRange") config.AuthKey = settings.get("AuthKey") config.LogMode = settings.get("LogMode") config.LogPath = settings.get("LogPath") config.LogDebug = settings.get("LogDebug") config.FrontPanelTheme = settings.get("FrontPanelTheme") config.ColorMode = settings.get("ColorMode") local cfv = util.new_validator() cfv.assert_type_num(config.SpeakerVolume) cfv.assert_range(config.SpeakerVolume, 0, 3) cfv.assert_channel(config.SVR_Channel) cfv.assert_channel(config.RTU_Channel) cfv.assert_type_num(config.ConnTimeout) cfv.assert_min(config.ConnTimeout, 2) cfv.assert_type_num(config.TrustedRange) cfv.assert_min(config.TrustedRange, 0) cfv.assert_type_str(config.AuthKey) if type(config.AuthKey) == "string" then local len = string.len(config.AuthKey) cfv.assert(len == 0 or len >= 8) end cfv.assert_type_int(config.LogMode) cfv.assert_range(config.LogMode, 0, 1) cfv.assert_type_str(config.LogPath) cfv.assert_type_bool(config.LogDebug) cfv.assert_type_int(config.FrontPanelTheme) cfv.assert_range(config.FrontPanelTheme, 1, 2) cfv.assert_type_int(config.ColorMode) cfv.assert_range(config.ColorMode, 1, themes.COLOR_MODE.NUM_MODES) cfv.assert_type_table(config.Peripherals) cfv.assert_type_table(config.Redstone) return cfv.valid() end -- create a new RTU unit
-- if this is for a PPM peripheral, auto fault clearing MUST stay enabled once access begins ---@nodiscard ---@param device table|nil peripheral device, if applicable function rtu.init_unit(device) local self = { discrete_inputs = {}, coils = {}, input_regs = {}, holding_regs = {}, io_count_cache = { 0, 0, 0, 0 } } local insert = table.insert local stub = function () log.warning("tried to call an RTU function stub") end ---@class rtu_device local public = {} ---@class rtu local protected = {} -- function to check if the peripheral (if exists) is faulted local function _is_faulted() return false end if device then _is_faulted = device.__p_is_faulted end -- refresh IO count local function _count_io() self.io_count_cache = { #self.discrete_inputs, #self.coils, #self.input_regs, #self.holding_regs } end -- return IO count ---@return integer discrete_inputs, integer coils, integer input_regs, integer holding_regs function public.io_count() return self.io_count_cache[1], self.io_count_cache[2], self.io_count_cache[3], self.io_count_cache[4] end -- pass a function through or generate one to call a function by name from the device ---@param f function|string function or device function name local function _as_func(f) if type(f) == "string" then local name = f if device then f = function (...) return device[name](...) end else f = stub end end return f end -- discrete inputs: single bit read-only -- connect discrete input ---@param f function|string function or function name ---@return integer count count of discrete inputs function protected.connect_di(f) insert(self.discrete_inputs, { read = _as_func(f) }) _count_io() return #self.discrete_inputs end -- read discrete input ---@param di_addr integer ---@return any value, boolean access_fault function public.read_di(di_addr) local value = self.discrete_inputs[di_addr].read() return value, _is_faulted() end -- coils: single bit read-write -- connect coil ---@param f_read function|string function or function name ---@param f_write function|string function or function name ---@return integer count count of coils function protected.connect_coil(f_read, f_write) insert(self.coils, { read = _as_func(f_read), write = _as_func(f_write) }) _count_io() return #self.coils end -- read coil ---@param coil_addr integer ---@return any value, boolean access_fault function public.read_coil(coil_addr) local value = self.coils[coil_addr].read() return value, _is_faulted() end -- write coil ---@param coil_addr integer ---@param value any ---@return boolean access_fault function public.write_coil(coil_addr, value) self.coils[coil_addr].write(value) return _is_faulted() end -- input registers: multi-bit read-only -- connect input register ---@param f function|string function or function name ---@return integer count count of input registers function protected.connect_input_reg(f) insert(self.input_regs, { read = _as_func(f) }) _count_io() return #self.input_regs end -- read input register ---@param reg_addr integer ---@return any value, boolean access_fault function public.read_input_reg(reg_addr) local value = self.input_regs[reg_addr].read() return value, _is_faulted() end -- holding registers: multi-bit read-write -- connect holding register ---@param f_read function|string function or function name ---@param f_write function|string function or function name ---@return integer count count of holding registers function protected.connect_holding_reg(f_read, f_write) insert(self.holding_regs, { read = _as_func(f_read), write = _as_func(f_write) }) _count_io() return #self.holding_regs end -- read holding register ---@param reg_addr integer ---@return any value, boolean access_fault function public.read_holding_reg(reg_addr) local value = self.holding_regs[reg_addr].read() return value, _is_faulted() end -- write holding register ---@param reg_addr integer ---@param value any ---@return boolean access_fault function public.write_holding_reg(reg_addr, value) self.holding_regs[reg_addr].write(value) return _is_faulted() end -- public RTU device access -- get the public interface to this RTU function protected.interface() return public end return protected end -- create an alarm speaker sounder ---@param speaker table device peripheral function rtu.init_sounder(speaker) ---@class rtu_speaker_sounder local spkr_ctl = { speaker = speaker, name = ppm.get_iface(speaker), playing = false, stream = audio.new_stream(), play = function () end, stop = function () end, continue = function () end } -- continue audio stream if playing function spkr_ctl.continue() if spkr_ctl.playing then if spkr_ctl.speaker ~= nil and spkr_ctl.stream.has_next_block() then local success = spkr_ctl.speaker.playAudio(spkr_ctl.stream.get_next_block(), config.SpeakerVolume) if not success then log.error(util.c("rtu_sounder(", spkr_ctl.name, "): error playing audio")) end end end end -- start audio stream playback function spkr_ctl.play() if not spkr_ctl.playing then spkr_ctl.playing = true return spkr_ctl.continue() end end -- stop audio stream playback function spkr_ctl.stop() spkr_ctl.playing = false spkr_ctl.speaker.stop() spkr_ctl.stream.stop() end return spkr_ctl end -- RTU Communications ---@nodiscard ---@param version string RTU version ---@param nic nic network interface device ---@param conn_watchdog watchdog watchdog reference function rtu.comms(version, nic, conn_watchdog) local self = { sv_addr = comms.BROADCAST, seq_num = util.time_ms() * 10, -- unique per peer, restarting will not re-use seq nums due to message rate r_seq_num = nil, ---@type nil|integer txn_id = 0, last_est_ack = ESTABLISH_ACK.ALLOW } local insert = table.insert comms.set_trusted_range(config.TrustedRange) -- PRIVATE FUNCTIONS -- -- configure modem channels nic.closeAll() nic.open(config.RTU_Channel) -- send a scada management packet ---@param msg_type MGMT_TYPE ---@param msg table local function _send(msg_type, msg) local s_pkt = comms.scada_packet() local m_pkt = comms.mgmt_packet() m_pkt.make(msg_type, msg) s_pkt.make(self.sv_addr, self.seq_num, PROTOCOL.SCADA_MGMT, m_pkt.raw_sendable()) nic.transmit(config.SVR_Channel, config.RTU_Channel, s_pkt) self.seq_num = self.seq_num + 1 end -- keep alive ack ---@param srv_time integer local function _send_keep_alive_ack(srv_time) _send(MGMT_TYPE.KEEP_ALIVE, { srv_time, util.time() }) end -- generate device advertisement table ---@nodiscard ---@param units table ---@return table advertisement local function _generate_advertisement(units) local advertisement = {} for i = 1, #units do local unit = units[i] ---@type rtu_unit_registry_entry if unit.type ~= nil then local advert = { unit.type, unit.index, unit.reactor } if unit.type == RTU_UNIT_TYPE.REDSTONE then insert(advert, unit.device) end insert(advertisement, advert) end end return advertisement end -- PUBLIC FUNCTIONS -- ---@class rtu_comms local public = {} -- send a MODBUS TCP packet ---@param m_pkt modbus_packet function public.send_modbus(m_pkt) local s_pkt = comms.scada_packet() s_pkt.make(self.sv_addr, self.seq_num, PROTOCOL.MODBUS_TCP, m_pkt.raw_sendable()) nic.transmit(config.SVR_Channel, config.RTU_Channel, s_pkt) self.seq_num = self.seq_num + 1 end -- unlink from the server ---@param rtu_state rtu_state function public.unlink(rtu_state) rtu_state.linked = false self.sv_addr = comms.BROADCAST self.r_seq_num = nil databus.tx_link_state(types.PANEL_LINK_STATE.DISCONNECTED) end -- close the connection to the server ---@param rtu_state rtu_state function public.close(rtu_state) conn_watchdog.cancel() public.unlink(rtu_state) _send(MGMT_TYPE.CLOSE, {}) end -- send establish request (includes advertisement) ---@param units table function public.send_establish(units) _send(MGMT_TYPE.ESTABLISH, { comms.version, version, DEVICE_TYPE.RTU, _generate_advertisement(units) }) end -- send capability advertisement ---@param units table function public.send_advertisement(units) _send(MGMT_TYPE.RTU_ADVERT, _generate_advertisement(units)) end -- notify that a peripheral was remounted ---@param unit_index integer RTU unit ID function public.send_remounted(unit_index) _send(MGMT_TYPE.RTU_DEV_REMOUNT, { unit_index }) end -- parse a MODBUS/SCADA packet ---@nodiscard ---@param side string ---@param sender integer ---@param reply_to integer ---@param message any ---@param distance integer ---@return modbus_frame|mgmt_frame|nil packet function public.parse_packet(side, sender, reply_to, message, distance) local s_pkt = nic.receive(side, sender, reply_to, message, distance) local pkt = nil if s_pkt then -- get as MODBUS TCP packet if s_pkt.protocol() == PROTOCOL.MODBUS_TCP then local m_pkt = comms.modbus_packet() if m_pkt.decode(s_pkt) then pkt = m_pkt.get() end -- get as SCADA management packet elseif s_pkt.protocol() == PROTOCOL.SCADA_MGMT then local mgmt_pkt = comms.mgmt_packet() if mgmt_pkt.decode(s_pkt) then pkt = mgmt_pkt.get() end else log.debug("illegal packet type " .. s_pkt.protocol(), true) end end return pkt end -- handle a MODBUS/SCADA packet ---@param packet modbus_frame|mgmt_frame ---@param units table RTU units ---@param rtu_state rtu_state ---@param sounders table speaker alarm sounders function public.handle_packet(packet, units, rtu_state, sounders) -- print a log message to the terminal as long as the UI isn't running local function println_ts(message) if not rtu_state.fp_ok then util.println_ts(message) end end local protocol = packet.scada_frame.protocol() local l_chan = packet.scada_frame.local_channel() local src_addr = packet.scada_frame.src_addr() if l_chan == config.RTU_Channel then -- check sequence number if self.r_seq_num == nil then self.r_seq_num = packet.scada_frame.seq_num() + 1 elseif self.r_seq_num ~= packet.scada_frame.seq_num() then log.warning("sequence out-of-order: last = " .. self.r_seq_num .. ", new = " .. packet.scada_frame.seq_num()) return elseif rtu_state.linked and (src_addr ~= self.sv_addr) then log.debug("received packet from unknown computer " .. src_addr .. " while linked (expected " .. self.sv_addr .. "); channel in use by another system?") return else self.r_seq_num = packet.scada_frame.seq_num() + 1 end -- feed watchdog on valid sequence number conn_watchdog.feed() -- handle packet if protocol == PROTOCOL.MODBUS_TCP then ---@cast packet modbus_frame if rtu_state.linked then local return_code ---@type boolean local reply ---@type modbus_packet -- handle MODBUS instruction if packet.unit_id <= #units then local unit = units[packet.unit_id] ---@type rtu_unit_registry_entry local unit_dbg_tag = " (unit " .. packet.unit_id .. ")" if unit.name == "redstone_io" then -- immediately execute redstone RTU requests return_code, reply = unit.modbus_io.handle_packet(packet) if not return_code then log.warning("requested MODBUS operation failed" .. unit_dbg_tag) end else -- check validity then pass off to unit comms thread return_code, reply = unit.modbus_io.check_request(packet) if return_code then -- check if there are more than 3 active transactions -- still queue the packet, but this may indicate a problem if unit.pkt_queue.length() > 3 then reply = modbus.reply__srv_device_busy(packet) log.debug("queueing new request with " .. unit.pkt_queue.length() .. " transactions already in the queue" .. unit_dbg_tag) end -- always queue the command even if busy unit.pkt_queue.push_packet(packet) else log.warning("cannot perform requested MODBUS operation" .. unit_dbg_tag) end end else -- unit ID out of range? reply = modbus.reply__gw_unavailable(packet) log.debug("received MODBUS packet for non-existent unit") end public.send_modbus(reply) else log.debug("discarding MODBUS packet before linked") end elseif protocol == PROTOCOL.SCADA_MGMT then ---@cast packet mgmt_frame -- SCADA management packet if rtu_state.linked then if packet.type == MGMT_TYPE.KEEP_ALIVE then -- keep alive request received, echo back if packet.length == 1 and type(packet.data[1]) == "number" then local timestamp = packet.data[1] local trip_time = util.time() - timestamp if trip_time > 750 then log.warning("RTU KEEP_ALIVE trip time > 750ms (" .. trip_time .. "ms)") end -- log.debug("RTU RTT = " .. trip_time .. "ms") _send_keep_alive_ack(timestamp) else log.debug("SCADA_MGMT keep alive packet length/type mismatch") end elseif packet.type == MGMT_TYPE.CLOSE then -- close connection conn_watchdog.cancel() public.unlink(rtu_state) println_ts("server connection closed by remote host") log.warning("server connection closed by remote host") elseif packet.type == MGMT_TYPE.RTU_ADVERT then -- request for capabilities again public.send_advertisement(units) elseif packet.type == MGMT_TYPE.RTU_TONE_ALARM then -- alarm tone update from supervisor if (packet.length == 1) and type(packet.data[1] == "table") and (#packet.data[1] == 8) then local states = packet.data[1] for i = 1, #sounders do local s = sounders[i] ---@type rtu_speaker_sounder -- set tone states for id = 1, #states do s.stream.set_active(id, states[id] == true) end end end else -- not supported log.debug("received unsupported SCADA_MGMT message type " .. packet.type) end elseif packet.type == MGMT_TYPE.ESTABLISH then if packet.length == 1 then local est_ack = packet.data[1] if est_ack == ESTABLISH_ACK.ALLOW then -- establish allowed rtu_state.linked = true self.sv_addr = packet.scada_frame.src_addr() println_ts("supervisor connection established") log.info("supervisor connection established") else -- establish denied if est_ack ~= self.last_est_ack then if est_ack == ESTABLISH_ACK.BAD_VERSION then -- version mismatch println_ts("supervisor comms version mismatch (try updating), retrying...") log.warning("supervisor connection denied due to comms version mismatch, retrying") else println_ts("supervisor connection denied, retrying...") log.warning("supervisor connection denied, retrying") end end public.unlink(rtu_state) end self.last_est_ack = est_ack -- report link state databus.tx_link_state(est_ack + 1) else log.debug("SCADA_MGMT establish packet length mismatch") end else log.debug("discarding non-link SCADA_MGMT packet before linked") end else -- should be unreachable assuming packet is from parse_packet() log.error("illegal packet type " .. protocol, true) end else log.debug("received packet on unconfigured channel " .. l_chan, true) end end return public end return rtu