cc-mek-scada/rtu/rtu.lua

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<br>
-- 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)
        self.r_seq_num = nil
        _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: next = " .. 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

                            -- unlink
                            self.sv_addr = comms.BROADCAST
                            rtu_state.linked = false
                        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