local log = require("scada-common.log") local mqueue = require("scada-common.mqueue") local ppm = require("scada-common.ppm") local tcd = require("scada-common.tcd") local types = require("scada-common.types") local util = require("scada-common.util") local databus = require("rtu.databus") local modbus = require("rtu.modbus") local renderer = require("rtu.renderer") local rtu = require("rtu.rtu") local boilerv_rtu = require("rtu.dev.boilerv_rtu") local dynamicv_rtu = require("rtu.dev.dynamicv_rtu") local envd_rtu = require("rtu.dev.envd_rtu") local imatrix_rtu = require("rtu.dev.imatrix_rtu") local sna_rtu = require("rtu.dev.sna_rtu") local sps_rtu = require("rtu.dev.sps_rtu") local turbinev_rtu = require("rtu.dev.turbinev_rtu") local core = require("graphics.core") local threads = {} local RTU_UNIT_TYPE = types.RTU_UNIT_TYPE local UNIT_HW_STATE = databus.RTU_UNIT_HW_STATE local MAIN_CLOCK = 0.5 -- (2Hz, 10 ticks) local COMMS_SLEEP = 100 -- (100ms, 2 ticks) ---@param smem rtu_shared_memory ---@param println_ts function ---@param iface string ---@param type string ---@param device table ---@param unit rtu_unit_registry_entry local function handle_unit_mount(smem, println_ts, iface, type, device, unit) local sys = smem.rtu_sys -- find disconnected device to reconnect -- note: cannot check isFormed as that would yield this coroutine and consume events if unit.name == iface then local resend_advert = false local faulted = false local unknown = false local invalid = false -- found, re-link unit.device = device if unit.type == RTU_UNIT_TYPE.VIRTUAL then resend_advert = true if type == "boilerValve" then -- boiler multiblock if unit.reactor < 1 or unit.reactor > 4 then invalid = true log.error(util.c("boiler '", unit.name, "' cannot init, not assigned to a valid unit in config")) end if (unit.index == false) or unit.index < 1 or unit.index > 2 then invalid = true log.error(util.c("boiler '", unit.name, "' cannot init, invalid index provided in config")) end unit.type = RTU_UNIT_TYPE.BOILER_VALVE elseif type == "turbineValve" then -- turbine multiblock if unit.reactor < 1 or unit.reactor > 4 then invalid = true log.error(util.c("turbine '", unit.name, "' cannot init, not assigned to a valid unit in config")) end if (unit.index == false) or unit.index < 1 or unit.index > 3 then invalid = true log.error(util.c("turbine '", unit.name, "' cannot init, invalid index provided in config")) end unit.type = RTU_UNIT_TYPE.TURBINE_VALVE elseif type == "dynamicValve" then -- dynamic tank multiblock if unit.reactor < 0 or unit.reactor > 4 then invalid = true log.error(util.c("dynamic tank '", unit.name, "' cannot init, no valid assignment provided in config")) end if (unit.reactor == 0 and ((unit.index == false) or unit.index < 1 or unit.index > 4)) or (unit.reactor > 0 and unit.index ~= 1) then invalid = true log.error(util.c("dynamic tank '", unit.name, "' cannot init, invalid index provided in config")) end unit.type = RTU_UNIT_TYPE.DYNAMIC_VALVE elseif type == "inductionPort" then -- induction matrix multiblock if unit.reactor ~= 0 then invalid = true log.error(util.c("induction matrix '", unit.name, "' cannot init, not assigned to facility in config")) end unit.type = RTU_UNIT_TYPE.IMATRIX elseif type == "spsPort" then -- SPS multiblock if unit.reactor ~= 0 then invalid = true log.error(util.c("SPS '", unit.name, "' cannot init, not assigned to facility in config")) end unit.type = RTU_UNIT_TYPE.SPS elseif type == "solarNeutronActivator" then -- SNA if unit.reactor < 1 or unit.reactor > 4 then invalid = true log.error(util.c("SNA '", unit.name, "' cannot init, not assigned to a valid unit in config")) end unit.type = RTU_UNIT_TYPE.SNA elseif type == "environmentDetector" then -- advanced peripherals environment detector if unit.reactor < 0 or unit.reactor > 4 then invalid = true log.error(util.c("environment detector '", unit.name, "' cannot init, no valid assignment provided in config")) end unit.type = RTU_UNIT_TYPE.ENV_DETECTOR else resend_advert = false log.error(util.c("virtual device '", unit.name, "' cannot init to an unknown type (", type, ")")) end databus.tx_unit_hw_type(unit.uid, unit.type) end -- if disconnected on startup, config wouldn't have been validated -- checking now that it has connected; the config isn't valid, so don't connect it if invalid then unit.hw_state = UNIT_HW_STATE.OFFLINE databus.tx_unit_hw_status(unit.uid, unit.hw_state) return end -- note for multiblock structures: if not formed, indexing the multiblock functions results in a PPM fault if unit.type == RTU_UNIT_TYPE.BOILER_VALVE then unit.rtu, faulted = boilerv_rtu.new(device) unit.formed = util.trinary(faulted, false, nil) elseif unit.type == RTU_UNIT_TYPE.TURBINE_VALVE then unit.rtu, faulted = turbinev_rtu.new(device) unit.formed = util.trinary(faulted, false, nil) elseif unit.type == RTU_UNIT_TYPE.DYNAMIC_VALVE then unit.rtu, faulted = dynamicv_rtu.new(device) unit.formed = util.trinary(faulted, false, nil) elseif unit.type == RTU_UNIT_TYPE.IMATRIX then unit.rtu, faulted = imatrix_rtu.new(device) unit.formed = util.trinary(faulted, false, nil) elseif unit.type == RTU_UNIT_TYPE.SPS then unit.rtu, faulted = sps_rtu.new(device) unit.formed = util.trinary(faulted, false, nil) elseif unit.type == RTU_UNIT_TYPE.SNA then unit.rtu, faulted = sna_rtu.new(device) elseif unit.type == RTU_UNIT_TYPE.ENV_DETECTOR then unit.rtu, faulted = envd_rtu.new(device) else unknown = true log.error(util.c("failed to identify reconnected RTU unit type (", unit.name, ")"), true) end if unit.is_multiblock then unit.hw_state = UNIT_HW_STATE.UNFORMED if unit.formed == false then log.info(util.c("assuming ", unit.name, " is not formed due to PPM faults while initializing")) end elseif faulted then unit.hw_state = UNIT_HW_STATE.FAULTED elseif not unknown then unit.hw_state = UNIT_HW_STATE.OK else unit.hw_state = UNIT_HW_STATE.OFFLINE end databus.tx_unit_hw_status(unit.uid, unit.hw_state) if not unknown then unit.modbus_io = modbus.new(unit.rtu, true) local type_name = types.rtu_type_to_string(unit.type) local message = util.c("reconnected the ", type_name, " on interface ", unit.name) println_ts(message) log.info(message) if resend_advert then sys.rtu_comms.send_advertisement(sys.units) else sys.rtu_comms.send_remounted(unit.uid) end end end end -- main thread ---@nodiscard ---@param smem rtu_shared_memory function threads.thread__main(smem) -- print a log message to the terminal as long as the UI isn't running local function println_ts(message) if not smem.rtu_state.fp_ok then util.println_ts(message) end end ---@class parallel_thread local public = {} -- execute thread function public.exec() databus.tx_rt_status("main", true) log.debug("main thread start") -- main loop clock local loop_clock = util.new_clock(MAIN_CLOCK) -- load in from shared memory local rtu_state = smem.rtu_state local sounders = smem.rtu_dev.sounders local nic = smem.rtu_sys.nic local rtu_comms = smem.rtu_sys.rtu_comms local conn_watchdog = smem.rtu_sys.conn_watchdog local units = smem.rtu_sys.units -- start unlinked (in case of restart) rtu_comms.unlink(rtu_state) -- start clock loop_clock.start() -- event loop while true do local event, param1, param2, param3, param4, param5 = util.pull_event() if event == "timer" and loop_clock.is_clock(param1) then -- blink heartbeat indicator databus.heartbeat() -- update speaker states for _, sounder in pairs(sounders) do -- re-compute output if needed, then play audio if available if sounder.stream.is_recompute_needed() then sounder.stream.compute_buffer() if sounder.stream.any_active() then sounder.play() else sounder.stop() end end end -- start next clock timer loop_clock.start() -- period tick, if we are not linked send establish request if not rtu_state.linked then -- advertise units rtu_comms.send_establish(units) end elseif event == "modem_message" then -- got a packet local packet = rtu_comms.parse_packet(param1, param2, param3, param4, param5) if packet ~= nil then -- pass the packet onto the comms message queue smem.q.mq_comms.push_packet(packet) end elseif event == "timer" and conn_watchdog.is_timer(param1) then -- haven't heard from server recently? close connection rtu_comms.close(rtu_state) elseif event == "timer" then -- notify timer callback dispatcher if no other timer case claimed this event tcd.handle(param1) elseif event == "peripheral_detach" then -- handle loss of a device local type, device = ppm.handle_unmount(param1) if type ~= nil and device ~= nil then if type == "modem" then -- we only care if this is our wireless modem if nic.is_modem(device) then nic.disconnect() println_ts("wireless modem disconnected!") log.warning("comms modem disconnected") local other_modem = ppm.get_wireless_modem() if other_modem then log.info("found another wireless modem, using it for comms") nic.connect(other_modem) else databus.tx_hw_modem(false) end else log.warning("non-comms modem disconnected") end elseif type == "speaker" then for i = 1, #sounders do if sounders[i].speaker == device then table.remove(sounders, i) log.warning(util.c("speaker ", param1, " disconnected")) println_ts("speaker disconnected") databus.tx_hw_spkr_count(#sounders) break end end else for i = 1, #units do -- find disconnected device if units[i].device == device then -- will let the PPM prevent crashes, which will indicate failures in MODBUS queries local unit = units[i] ---@type rtu_unit_registry_entry local type_name = types.rtu_type_to_string(unit.type) println_ts(util.c("lost the ", type_name, " on interface ", unit.name)) log.warning(util.c("lost the ", type_name, " unit peripheral on interface ", unit.name)) unit.hw_state = UNIT_HW_STATE.OFFLINE databus.tx_unit_hw_status(unit.uid, unit.hw_state) break end end end end elseif event == "peripheral" then -- peripheral connect local type, device = ppm.mount(param1) if type ~= nil and device ~= nil then if type == "modem" then if device.isWireless() and not nic.is_connected() then -- reconnected modem nic.connect(device) println_ts("wireless modem reconnected.") log.info("comms modem reconnected") databus.tx_hw_modem(true) elseif device.isWireless() then log.info("unused wireless modem reconnected") else log.info("wired modem reconnected") end elseif type == "speaker" then table.insert(sounders, rtu.init_sounder(device)) println_ts("speaker connected") log.info(util.c("connected speaker ", param1)) databus.tx_hw_spkr_count(#sounders) else -- relink lost peripheral to correct unit entry for i = 1, #units do handle_unit_mount(smem, println_ts, param1, type, device, units[i]) end end end elseif event == "mouse_click" or event == "mouse_up" or event == "mouse_drag" or event == "mouse_scroll" or event == "double_click" then -- handle a mouse event renderer.handle_mouse(core.events.new_mouse_event(event, param1, param2, param3)) elseif event == "speaker_audio_empty" then -- handle empty speaker audio buffer for i = 1, #sounders do local sounder = sounders[i] ---@type rtu_speaker_sounder if sounder.name == param1 then sounder.continue() break end end end -- check for termination request if event == "terminate" or ppm.should_terminate() then rtu_state.shutdown = true log.info("terminate requested, main thread exiting") break end end end -- execute the thread in a protected mode, retrying it on return if not shutting down function public.p_exec() local rtu_state = smem.rtu_state while not rtu_state.shutdown do local status, result = pcall(public.exec) if status == false then log.fatal(util.strval(result)) end databus.tx_rt_status("main", false) if not rtu_state.shutdown then log.info("main thread restarting in 5 seconds...") util.psleep(5) end end end return public end -- communications handler thread ---@nodiscard ---@param smem rtu_shared_memory function threads.thread__comms(smem) ---@class parallel_thread local public = {} -- execute thread function public.exec() databus.tx_rt_status("comms", true) log.debug("comms thread start") -- load in from shared memory local rtu_state = smem.rtu_state local sounders = smem.rtu_dev.sounders local rtu_comms = smem.rtu_sys.rtu_comms local units = smem.rtu_sys.units local comms_queue = smem.q.mq_comms local last_update = util.time() -- thread loop while true do local handle_start = util.time() -- check for messages in the message queue while not shut down while comms_queue.ready() and not rtu_state.shutdown do local msg = comms_queue.pop() if msg ~= nil then if msg.qtype == mqueue.TYPE.COMMAND then -- received a command elseif msg.qtype == mqueue.TYPE.DATA then -- received data elseif msg.qtype == mqueue.TYPE.PACKET then -- received a packet -- handle the packet (rtu_state passed to allow setting link flag, sounders passed to manage alarm audio) rtu_comms.handle_packet(msg.message, units, rtu_state, sounders) end end -- max 100ms spent processing queue if util.time() - handle_start > 100 then log.warning("comms thread exceeded 100ms queue process limit") break end end -- quick yield util.nop() -- check for termination request if rtu_state.shutdown then rtu_comms.close(rtu_state) log.info("comms thread exiting") break end -- delay before next check last_update = util.adaptive_delay(COMMS_SLEEP, last_update) end end -- execute the thread in a protected mode, retrying it on return if not shutting down function public.p_exec() local rtu_state = smem.rtu_state while not rtu_state.shutdown do local status, result = pcall(public.exec) if status == false then log.fatal(util.strval(result)) end databus.tx_rt_status("comms", false) if not rtu_state.shutdown then log.info("comms thread restarting in 5 seconds...") util.psleep(5) end end end return public end -- per-unit communications handler thread ---@nodiscard ---@param smem rtu_shared_memory ---@param unit rtu_unit_registry_entry function threads.thread__unit_comms(smem, unit) ---@class parallel_thread local public = {} -- execute thread function public.exec() databus.tx_rt_status("unit_" .. unit.uid, true) log.debug(util.c("rtu unit thread start -> ", types.rtu_type_to_string(unit.type), " (", unit.name, ")")) -- load in from shared memory local rtu_state = smem.rtu_state local rtu_comms = smem.rtu_sys.rtu_comms local packet_queue = unit.pkt_queue local last_update = util.time() local last_f_check = 0 local detail_name = util.c(types.rtu_type_to_string(unit.type), " (", unit.name, ") [", unit.index, "] for reactor ", unit.reactor) local short_name = util.c(types.rtu_type_to_string(unit.type), " (", unit.name, ")") if packet_queue == nil then log.error("rtu unit thread created without a message queue, exiting...", true) return end -- thread loop while true do -- check for messages in the message queue while packet_queue.ready() and not rtu_state.shutdown do local msg = packet_queue.pop() if msg ~= nil then if msg.qtype == mqueue.TYPE.COMMAND then -- received a command elseif msg.qtype == mqueue.TYPE.DATA then -- received data elseif msg.qtype == mqueue.TYPE.PACKET then -- received a packet local _, reply = unit.modbus_io.handle_packet(msg.message) rtu_comms.send_modbus(reply) end end -- quick yield util.nop() end -- check for termination request if rtu_state.shutdown then log.info("rtu unit thread exiting -> " .. short_name) break end -- check if multiblock is still formed if this is a multiblock if unit.is_multiblock and (util.time_ms() - last_f_check > 250) then local is_formed = unit.device.isFormed() last_f_check = util.time_ms() if unit.formed == nil then unit.formed = is_formed if is_formed then unit.hw_state = UNIT_HW_STATE.OK end end if not unit.formed then unit.hw_state = UNIT_HW_STATE.UNFORMED end if (not unit.formed) and is_formed then -- newly re-formed local iface = ppm.get_iface(unit.device) if iface then log.info(util.c("unmounting and remounting reformed RTU unit ", detail_name)) ppm.unmount(unit.device) local type, device = ppm.mount(iface) local faulted = false if device ~= nil then if type == "boilerValve" and unit.type == RTU_UNIT_TYPE.BOILER_VALVE then -- boiler multiblock unit.device = device unit.rtu, faulted = boilerv_rtu.new(device) unit.formed = device.isFormed() unit.modbus_io = modbus.new(unit.rtu, true) elseif type == "turbineValve" and unit.type == RTU_UNIT_TYPE.TURBINE_VALVE then -- turbine multiblock unit.device = device unit.rtu, faulted = turbinev_rtu.new(device) unit.formed = device.isFormed() unit.modbus_io = modbus.new(unit.rtu, true) elseif type == "dynamicValve" and unit.type == RTU_UNIT_TYPE.DYNAMIC_VALVE then -- dynamic tank multiblock unit.device = device unit.rtu, faulted = dynamicv_rtu.new(device) unit.formed = device.isFormed() unit.modbus_io = modbus.new(unit.rtu, true) elseif type == "inductionPort" and unit.type == RTU_UNIT_TYPE.IMATRIX then -- induction matrix multiblock unit.device = device unit.rtu, faulted = imatrix_rtu.new(device) unit.formed = device.isFormed() unit.modbus_io = modbus.new(unit.rtu, true) elseif type == "spsPort" and unit.type == RTU_UNIT_TYPE.SPS then -- SPS multiblock unit.device = device unit.rtu, faulted = sps_rtu.new(device) unit.formed = device.isFormed() unit.modbus_io = modbus.new(unit.rtu, true) else log.error("illegal remount of non-multiblock RTU or type change attempted for " .. short_name, true) end if unit.formed and faulted then -- something is still wrong = can't mark as formed yet unit.formed = false unit.hw_state = UNIT_HW_STATE.UNFORMED log.info(util.c("assuming ", unit.name, " is not formed due to PPM faults while initializing")) else unit.hw_state = UNIT_HW_STATE.OK rtu_comms.send_remounted(unit.uid) end local type_name = types.rtu_type_to_string(unit.type) log.info(util.c("reconnected the ", type_name, " on interface ", unit.name)) else -- fully lost the peripheral now :( log.error(util.c(unit.name, " lost (failed reconnect)")) end else log.error("failed to get interface of previously connected RTU unit " .. detail_name, true) end end unit.formed = is_formed end -- check hardware status if unit.device.__p_is_healthy() then if unit.hw_state == UNIT_HW_STATE.FAULTED then unit.hw_state = UNIT_HW_STATE.OK end else if unit.hw_state == UNIT_HW_STATE.OK then unit.hw_state = UNIT_HW_STATE.FAULTED end end -- update hw status databus.tx_unit_hw_status(unit.uid, unit.hw_state) -- delay before next check last_update = util.adaptive_delay(COMMS_SLEEP, last_update) end end -- execute the thread in a protected mode, retrying it on return if not shutting down function public.p_exec() local rtu_state = smem.rtu_state while not rtu_state.shutdown do local status, result = pcall(public.exec) if status == false then log.fatal(util.strval(result)) end databus.tx_rt_status("unit_" .. unit.uid, false) if not rtu_state.shutdown then log.info(util.c("rtu unit thread ", types.rtu_type_to_string(unit.type), " (", unit.name, ") restarting in 5 seconds...")) util.psleep(5) end end end return public end return threads