Minecraft/cc-mek-scada
1
0
Fork 0
mirror of https://github.com/MikaylaFischler/cc-mek-scada.git synced 2024-08-30 18:22:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

#9 MODBUS test code; fixed rtu, modbus, redstone_rtu, and rsio bugs

Browse Source
This commit is contained in:
Mikayla Fischler 2022-05-27 18:10:06 -04:00
parent 51111f707f
commit 6df0a1d149
8 changed files with 385 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
rtu/dev/redstone_rtu.lua
View File

@ -4,6 +4,7 @@ local rsio = require("scada-common.rsio")
local redstone_rtu = {}
local digital_read = rsio.digital_read
local digital_write = rsio.digital_write
local digital_is_active = rsio.digital_is_active
-- create new redstone device
@ -61,12 +62,11 @@ redstone_rtu.new = function ()
f_write = function (level)
local output = rs.getBundledOutput(side)
local active = digital_is_active(channel, level)
if active then
colors.combine(output, color)
if digital_write(channel, level) then
output = colors.combine(output, color)
else
colors.subtract(output, color)
output = colors.subtract(output, color)
end
rs.setBundledOutput(side, output)

35
rtu/modbus.lua
View File

@ -28,7 +28,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
local readings = {}
local access_fault = false
local _, coils, _, _ = self.rtu.io_count()
local return_ok = ((c_addr_start + count) <= coils) and (count > 0)
local return_ok = ((c_addr_start + count) <= (coils + 1)) and (count > 0)
if return_ok then
for i = 1, count do
@ -74,7 +74,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
local readings = {}
local access_fault = false
local discrete_inputs, _, _, _ = self.rtu.io_count()
local return_ok = ((di_addr_start + count) <= discrete_inputs) and (count > 0)
local return_ok = ((di_addr_start + count) <= (discrete_inputs + 1)) and (count > 0)
if return_ok then
for i = 1, count do
@ -120,7 +120,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
local readings = {}
local access_fault = false
local _, _, _, hold_regs = self.rtu.io_count()
local return_ok = ((hr_addr_start + count) <= hold_regs) and (count > 0)
local return_ok = ((hr_addr_start + count) <= (hold_regs + 1)) and (count > 0)
if return_ok then
for i = 1, count do
@ -166,7 +166,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
local readings = {}
local access_fault = false
local _, _, input_regs, _ = self.rtu.io_count()
local return_ok = ((ir_addr_start + count) <= input_regs) and (count > 0)
local return_ok = ((ir_addr_start + count) <= (input_regs + 1)) and (count > 0)
if return_ok then
for i = 1, count do
@ -255,7 +255,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
local response = nil
local _, coils, _, _ = self.rtu.io_count()
local count = #values
local return_ok = ((c_addr_start + count) <= coils) and (count > 0)
local return_ok = ((c_addr_start + count) <= (coils + 1)) and (count > 0)
if return_ok then
for i = 1, count do
@ -282,12 +282,12 @@ modbus.new = function (rtu_dev, use_parallel_read)
local response = nil
local _, _, _, hold_regs = self.rtu.io_count()
local count = #values
local return_ok = ((hr_addr_start + count) <= hold_regs) and (count > 0)
local return_ok = ((hr_addr_start + count) <= (hold_regs + 1)) and (count > 0)
if return_ok then
for i = 1, count do
local addr = hr_addr_start + i - 1
local access_fault = self.rtu.write_coil(addr, values[i])
local access_fault = self.rtu.write_holding_reg(addr, values[i])
if access_fault then
return_ok = false
@ -309,12 +309,12 @@ modbus.new = function (rtu_dev, use_parallel_read)
local return_code = true
local response = { MODBUS_EXCODE.ACKNOWLEDGE }
if #packet.data == 2 then
if packet.length == 2 then
-- handle by function code
if packet.func_code == MODBUS_FCODE.READ_COILS then
elseif packet.func_code == MODBUS_FCODE.READ_DISCRETE_INPUTS then
elseif packet.func_code == MODBUS_FCODE.READ_MUL_HOLD_REGS then
elseif packet.func_code == MODBUS_FCODE.READ_INPUT_REGISTERS then
elseif packet.func_code == MODBUS_FCODE.READ_INPUT_REGS then
elseif packet.func_code == MODBUS_FCODE.WRITE_SINGLE_COIL then
elseif packet.func_code == MODBUS_FCODE.WRITE_SINGLE_HOLD_REG then
elseif packet.func_code == MODBUS_FCODE.WRITE_MUL_COILS then
@ -351,7 +351,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
local return_code = true
local response = nil
if #packet.data == 2 then
if packet.length == 2 then
-- handle by function code
if packet.func_code == MODBUS_FCODE.READ_COILS then
return_code, response = _1_read_coils(packet.data[1], packet.data[2])
@ -359,7 +359,7 @@ modbus.new = function (rtu_dev, use_parallel_read)
return_code, response = _2_read_discrete_inputs(packet.data[1], packet.data[2])
elseif packet.func_code == MODBUS_FCODE.READ_MUL_HOLD_REGS then
return_code, response = _3_read_multiple_holding_registers(packet.data[1], packet.data[2])
elseif packet.func_code == MODBUS_FCODE.READ_INPUT_REGISTERS then
elseif packet.func_code == MODBUS_FCODE.READ_INPUT_REGS then
return_code, response = _4_read_input_registers(packet.data[1], packet.data[2])
elseif packet.func_code == MODBUS_FCODE.WRITE_SINGLE_COIL then
return_code, response = _5_write_single_coil(packet.data[1], packet.data[2])
@ -384,14 +384,13 @@ modbus.new = function (rtu_dev, use_parallel_read)
if not return_code then
-- echo back with error flag
func_code = bit.bor(packet.func_code, MODBUS_FCODE.ERROR_FLAG)
if type(response) == "nil" then
response = { }
elseif type(response) == "number" then
response = { response }
elseif type(response) == "table" then
response = response
end
if type(response) == "table" then
elseif type(response) == "nil" then
response = {}
else
response = { response }
end
-- create reply

12
rtu/rtu.lua
View File

@ -54,7 +54,7 @@ rtu.init_unit = function ()
---@param f function
---@return integer count count of discrete inputs
protected.connect_di = function (f)
insert(self.discrete_inputs, f)
insert(self.discrete_inputs, { read = f })
_count_io()
return #self.discrete_inputs
end
@ -64,7 +64,7 @@ rtu.init_unit = function ()
---@return any value, boolean access_fault
public.read_di = function (di_addr)
ppm.clear_fault()
local value = self.discrete_inputs[di_addr]()
local value = self.discrete_inputs[di_addr].read()
return value, ppm.is_faulted()
end
@ -105,7 +105,7 @@ rtu.init_unit = function ()
---@param f function
---@return integer count count of input registers
protected.connect_input_reg = function (f)
insert(self.input_regs, f)
insert(self.input_regs, { read = f })
_count_io()
return #self.input_regs
end
@ -115,7 +115,7 @@ rtu.init_unit = function ()
---@return any value, boolean access_fault
public.read_input_reg = function (reg_addr)
ppm.clear_fault()
local value = self.coils[reg_addr]()
local value = self.input_regs[reg_addr].read()
return value, ppm.is_faulted()
end
@ -136,7 +136,7 @@ rtu.init_unit = function ()
---@return any value, boolean access_fault
public.read_holding_reg = function (reg_addr)
ppm.clear_fault()
local value = self.coils[reg_addr].read()
local value = self.holding_regs[reg_addr].read()
return value, ppm.is_faulted()
end
@ -146,7 +146,7 @@ rtu.init_unit = function ()
---@return boolean access_fault
public.write_holding_reg = function (reg_addr, value)
ppm.clear_fault()
self.coils[reg_addr].write(value)
self.holding_regs[reg_addr].write(value)
return ppm.is_faulted()
end

2
rtu/startup.lua
View File

@ -24,7 +24,7 @@ local imatrix_rtu = require("rtu.dev.imatrix_rtu")
local turbine_rtu = require("rtu.dev.turbine_rtu")
local turbinev_rtu = require("rtu.dev.turbinev_rtu")
local RTU_VERSION = "alpha-v0.7.0"
local RTU_VERSION = "alpha-v0.7.1"
local rtu_t = types.rtu_t

58
scada-common/rsio.lua
View File

@ -110,55 +110,51 @@ end
local _B_AND = bit.band
local _TRINARY = function (cond, t, f) if cond then return t else return f end end
local _DI_ACTIVE_HIGH = function (level) return level == IO_LVL.HIGH end
local _DI_ACTIVE_LOW = function (level) return level == IO_LVL.LOW end
local _DO_ACTIVE_HIGH = function (on) return _TRINARY(on, IO_LVL.HIGH, IO_LVL.LOW) end
local _DO_ACTIVE_LOW = function (on) return _TRINARY(on, IO_LVL.LOW, IO_LVL.HIGH) end
local function _ACTIVE_HIGH(level) return level == IO_LVL.HIGH end
local function _ACTIVE_LOW(level) return level == IO_LVL.LOW end
-- I/O mappings to I/O function and I/O mode
local RS_DIO_MAP = {
-- F_SCRAM
{ _f = _DI_ACTIVE_LOW, mode = IO_DIR.IN },
{ _f = _ACTIVE_LOW, mode = IO_DIR.IN },
-- R_SCRAM
{ _f = _DI_ACTIVE_LOW, mode = IO_DIR.IN },
{ _f = _ACTIVE_LOW, mode = IO_DIR.IN },
-- R_ENABLE
{ _f = _DI_ACTIVE_HIGH, mode = IO_DIR.IN },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.IN },
-- F_ALARM
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- WASTE_PO
{ _f = _DO_ACTIVE_LOW, mode = IO_DIR.OUT },
{ _f = _ACTIVE_LOW, mode = IO_DIR.OUT },
-- WASTE_PU
{ _f = _DO_ACTIVE_LOW, mode = IO_DIR.OUT },
{ _f = _ACTIVE_LOW, mode = IO_DIR.OUT },
-- WASTE_AM
{ _f = _DO_ACTIVE_LOW, mode = IO_DIR.OUT },
{ _f = _ACTIVE_LOW, mode = IO_DIR.OUT },
-- R_ALARM
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_SCRAMMED
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_AUTO_SCRAM
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_ACTIVE
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_AUTO_CTRL
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_DMG_CRIT
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_HIGH_TEMP
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_NO_COOLANT
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_EXCESS_HC
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_EXCESS_WS
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_INSUFF_FUEL
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_PLC_FAULT
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT },
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT },
-- R_PLC_TIMEOUT
{ _f = _DO_ACTIVE_HIGH, mode = IO_DIR.OUT }
{ _f = _ACTIVE_HIGH, mode = IO_DIR.OUT }
}
-- get the mode of a channel
@ -244,13 +240,13 @@ end
-- returns the level corresponding to active
---@param channel RS_IO
---@param active boolean
---@return IO_LVL
rsio.digital_write = function (channel, active)
---@param level IO_LVL
---@return boolean
rsio.digital_write = function (channel, level)
if type(channel) ~= "number" or channel < RS_IO.F_ALARM or channel > RS_IO.R_PLC_TIMEOUT then
return IO_LVL.LOW
return false
else
return RS_DIO_MAP[channel]._f(active)
return RS_DIO_MAP[channel]._f(level)
end
end

236
test/modbustest.lua Normal file
View File

@ -0,0 +1,236 @@
require("/initenv").init_env()
local types = require("scada-common.types")
local util = require("scada-common.util")
local testutils = require("test.testutils")
local modbus = require("rtu.modbus")
local redstone_rtu = require("rtu.dev.redstone_rtu")
local rsio = require("scada-common.rsio")
local print = util.print
local println = util.println
local MODBUS_FCODE = types.MODBUS_FCODE
local MODBUS_EXCODE = types.MODBUS_EXCODE
println("starting redstone RTU and MODBUS tester")
println("")
-- RTU init --
print(">>> init redstone RTU: ")
local rs_rtu = redstone_rtu.new()
local di, c, ir, hr = rs_rtu.io_count()
assert(di == 0 and c == 0 and ir == 0 and hr == 0, "IOCOUNT_0")
rs_rtu.link_di("back", colors.black)
rs_rtu.link_di("back", colors.blue)
rs_rtu.link_do(rsio.IO.F_ALARM, "back", colors.red)
rs_rtu.link_do(rsio.IO.WASTE_AM, "back", colors.purple)
rs_rtu.link_ai("right")
rs_rtu.link_ao("left")
di, c, ir, hr = rs_rtu.io_count()
assert(di == 2, "IOCOUNT_DI")
assert(c == 2, "IOCOUNT_C")
assert(ir == 1, "IOCOUNT_IR")
assert(hr == 1, "IOCOUNT_HR")
println("OK")
-- MODBUS testing --
local rs_modbus = modbus.new(rs_rtu, false)
local mbt = testutils.modbus_tester(rs_modbus, MODBUS_FCODE.ERROR_FLAG)
-------------------------
--- CHECKING REQUESTS ---
-------------------------
println(">>> checking MODBUS requests:")
print("read c {0}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {0})
mbt.test_error__check_request(MODBUS_EXCODE.NEG_ACKNOWLEDGE)
println("PASS")
print("99 {1,2}: ")
mbt.pkt_set(99, {1, 2})
mbt.test_error__check_request(MODBUS_EXCODE.ILLEGAL_FUNCTION)
println("PASS")
print("read c {1,2}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {1, 2})
mbt.test_success__check_request(MODBUS_EXCODE.ACKNOWLEDGE)
println("PASS")
testutils.pause()
--------------------
--- BAD REQUESTS ---
--------------------
println(">>> trying bad requests:")
print("read di {1,10}: ")
mbt.pkt_set(MODBUS_FCODE.READ_DISCRETE_INPUTS, {1, 10})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read di {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_DISCRETE_INPUTS, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read di {1,0}: ")
mbt.pkt_set(MODBUS_FCODE.READ_DISCRETE_INPUTS, {1, 0})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read c {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read c {1,0}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {1, 0})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read ir {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_INPUT_REGS, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read ir {1,0}: ")
mbt.pkt_set(MODBUS_FCODE.READ_INPUT_REGS, {1, 0})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("read hr {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_MUL_HOLD_REGS, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("write c {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_COIL, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("write mul c {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_COIL, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("write mul c {5,{1}}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_COIL, {5, {1}})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("write hr {5,1}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_HOLD_REG, {5, 1})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
print("write mul hr {5,{1}}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_HOLD_REG, {5, {1}})
mbt.test_error__handle_packet(MODBUS_EXCODE.ILLEGAL_DATA_ADDR)
println("PASS")
testutils.pause()
----------------------
--- READING INPUTS ---
----------------------
println(">>> reading inputs:")
print("read di {1,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_DISCRETE_INPUTS, {1, 1})
mbt.test_success__handle_packet()
print("read di {2,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_DISCRETE_INPUTS, {2, 1})
mbt.test_success__handle_packet()
print("read di {1,2}: ")
mbt.pkt_set(MODBUS_FCODE.READ_DISCRETE_INPUTS, {1, 2})
mbt.test_success__handle_packet()
print("read ir {1,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_INPUT_REGS, {1, 1})
mbt.test_success__handle_packet()
testutils.pause()
-----------------------
--- WRITING OUTPUTS ---
-----------------------
println(">>> writing outputs:")
print("write mul c {1,{LOW,LOW}}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_MUL_COILS, {1, {rsio.IO_LVL.LOW, rsio.IO_LVL.LOW}})
mbt.test_success__handle_packet()
testutils.pause()
print("write c {1,HIGH}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_COIL, {1, rsio.IO_LVL.HIGH})
mbt.test_success__handle_packet()
testutils.pause()
print("write c {2,HIGH}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_COIL, {2, rsio.IO_LVL.HIGH})
mbt.test_success__handle_packet()
testutils.pause()
print("write hr {1,7}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_SINGLE_HOLD_REG, {1, 7})
mbt.test_success__handle_packet()
testutils.pause()
print("write mul hr {1,{4}}: ")
mbt.pkt_set(MODBUS_FCODE.WRITE_MUL_HOLD_REGS, {1, {4}})
mbt.test_success__handle_packet()
println("PASS")
testutils.pause()
-----------------------
--- READING OUTPUTS ---
-----------------------
println(">>> reading outputs:")
print("read c {1,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {1, 1})
mbt.test_success__handle_packet()
print("read c {2,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {2, 1})
mbt.test_success__handle_packet()
print("read c {1,2}: ")
mbt.pkt_set(MODBUS_FCODE.READ_COILS, {1, 2})
mbt.test_success__handle_packet()
print("read hr {1,1}: ")
mbt.pkt_set(MODBUS_FCODE.READ_MUL_HOLD_REGS, {1, 1})
mbt.test_success__handle_packet()
println("PASS")
println("TEST COMPLETE")

36
test/rstest.lua
View File

@ -47,8 +47,7 @@ end
assert(max_value == cid, "RS_IO last IDx out-of-sync with count: " .. max_value .. " (count " .. cid .. ")")
---@diagnostic disable-next-line: undefined-field
os.sleep(1)
testutils.pause()
println(">>> checking invalid channels:")
@ -57,8 +56,7 @@ testutils.test_func_nil("rsio.to_string", rsio.to_string, "")
testutils.test_func("rsio.get_io_mode", rsio.get_io_mode, { -1, 100, false }, IO_MODE.ANALOG_IN)
testutils.test_func_nil("rsio.get_io_mode", rsio.get_io_mode, IO_MODE.ANALOG_IN)
---@diagnostic disable-next-line: undefined-field
os.sleep(1)
testutils.pause()
println(">>> checking validity checks:")
@ -76,8 +74,7 @@ testutils.test_func("rsio.is_color", rsio.is_color, ic_t_list, true)
testutils.test_func("rsio.is_color", rsio.is_color, { 0, 999999, colors.combine(colors.red, colors.blue, colors.black) }, false)
testutils.test_func_nil("rsio.is_color", rsio.is_color, false)
---@diagnostic disable-next-line: undefined-field
os.sleep(1)
testutils.pause()
println(">>> checking channel-independent I/O wrappers:")
@ -98,8 +95,7 @@ assert(rsio.analog_write(4, 0, 15) == 4, "RS_WRITE_4_15")
assert(rsio.analog_write(12, 0, 15) == 12, "RS_WRITE_12_15")
println("PASS")
---@diagnostic disable-next-line: undefined-field
os.sleep(1)
testutils.pause()
println(">>> checking channel I/O:")
@ -124,25 +120,25 @@ println("PASS")
print("rsio.digital_write(...): ")
-- check output channels
assert(rsio.digital_write(IO.F_ALARM, false) == IO_LVL.LOW, "IO_F_ALARM_FALSE")
assert(rsio.digital_write(IO.F_ALARM, true) == IO_LVL.HIGH, "IO_F_ALARM_TRUE")
assert(rsio.digital_write(IO.WASTE_PO, false) == IO_LVL.HIGH, "IO_WASTE_PO_FALSE")
assert(rsio.digital_write(IO.WASTE_PO, true) == IO_LVL.LOW, "IO_WASTE_PO_TRUE")
assert(rsio.digital_write(IO.WASTE_PU, false) == IO_LVL.HIGH, "IO_WASTE_PU_FALSE")
assert(rsio.digital_write(IO.WASTE_PU, true) == IO_LVL.LOW, "IO_WASTE_PU_TRUE")
assert(rsio.digital_write(IO.WASTE_AM, false) == IO_LVL.HIGH, "IO_WASTE_AM_FALSE")
assert(rsio.digital_write(IO.WASTE_AM, true) == IO_LVL.LOW, "IO_WASTE_AM_TRUE")
assert(rsio.digital_write(IO.F_ALARM, IO_LVL.LOW) == false, "IO_F_ALARM_FALSE")
assert(rsio.digital_write(IO.F_ALARM, IO_LVL.HIGH) == true, "IO_F_ALARM_TRUE")
assert(rsio.digital_write(IO.WASTE_PO, IO_LVL.HIGH) == false, "IO_WASTE_PO_FALSE")
assert(rsio.digital_write(IO.WASTE_PO, IO_LVL.LOW) == true, "IO_WASTE_PO_TRUE")
assert(rsio.digital_write(IO.WASTE_PU, IO_LVL.HIGH) == false, "IO_WASTE_PU_FALSE")
assert(rsio.digital_write(IO.WASTE_PU, IO_LVL.LOW) == true, "IO_WASTE_PU_TRUE")
assert(rsio.digital_write(IO.WASTE_AM, IO_LVL.HIGH) == false, "IO_WASTE_AM_FALSE")
assert(rsio.digital_write(IO.WASTE_AM, IO_LVL.LOW) == true, "IO_WASTE_AM_TRUE")
-- check all reactor output channels (all are active high)
for i = IO.R_ALARM, (IO.R_PLC_TIMEOUT - IO.R_ALARM + 1) do
assert(rsio.to_string(i) ~= "", "REACTOR_IO_BAD_CHANNEL")
assert(rsio.digital_write(i, false) == IO_LVL.LOW, "IO_" .. rsio.to_string(i) .. "_FALSE")
assert(rsio.digital_write(i, true) == IO_LVL.HIGH, "IO_" .. rsio.to_string(i) .. "_TRUE")
assert(rsio.digital_write(i, IO_LVL.LOW) == false, "IO_" .. rsio.to_string(i) .. "_FALSE")
assert(rsio.digital_write(i, IO_LVL.HIGH) == true, "IO_" .. rsio.to_string(i) .. "_TRUE")
end
-- non-outputs should always return false
assert(rsio.digital_write(IO.F_SCRAM, true) == IO_LVL.LOW, "IO_IN_WRITE_LOW")
assert(rsio.digital_write(IO.F_SCRAM, false) == IO_LVL.LOW, "IO_IN_WRITE_HIGH")
assert(rsio.digital_write(IO.F_SCRAM, IO_LVL.LOW) == false, "IO_IN_WRITE_LOW")
assert(rsio.digital_write(IO.F_SCRAM, IO_LVL.LOW) == false, "IO_IN_WRITE_HIGH")
println("PASS")

78
test/testutils.lua
View File

@ -41,4 +41,82 @@ function testutils.test_func_nil(name, f, result)
println("PASS")
end
-- get something as a string
---@param result any
---@return string
function testutils.stringify(result)
return textutils.serialize(result, { allow_repetitions = true, compact = true })
end
-- pause for 1 second, or the provided seconds
---@param seconds? number
function testutils.pause(seconds)
seconds = seconds or 1.0
---@diagnostic disable-next-line: undefined-field
os.sleep(seconds)
end
-- create a new MODBUS tester
---@param modbus modbus modbus object
---@param error_flag MODBUS_FCODE MODBUS_FCODE.ERROR_FLAG
function testutils.modbus_tester(modbus, error_flag)
-- test packet
---@type modbus_frame
local packet = {
txn_id = 0,
length = 0,
unit_id = 0,
func_code = 0,
data = {},
scada_frame = nil
}
---@class modbus_tester
local public = {}
-- set the packet function and data for the next test
---@param func MODBUS_FCODE function code
---@param data table
function public.pkt_set(func, data)
packet.length = #data
packet.data = data
packet.func_code = func
end
-- check the current packet, expecting an error
---@param excode MODBUS_EXCODE exception code to expect
function public.test_error__check_request(excode)
local rcode, reply = modbus.check_request(packet)
assert(rcode == false, "CHECK_NOT_FAIL")
assert(reply.get().func_code == bit.bor(packet.func_code, error_flag), "WRONG_FCODE")
assert(reply.get().data[1] == excode, "EXCODE_MISMATCH")
end
-- test the current packet, expecting an error
---@param excode MODBUS_EXCODE exception code to expect
function public.test_error__handle_packet(excode)
local rcode, reply = modbus.handle_packet(packet)
assert(rcode == false, "CHECK_NOT_FAIL")
assert(reply.get().func_code == bit.bor(packet.func_code, error_flag), "WRONG_FCODE")
assert(reply.get().data[1] == excode, "EXCODE_MISMATCH")
end
-- check the current packet, expecting success
---@param excode MODBUS_EXCODE exception code to expect
function public.test_success__check_request(excode)
local rcode, reply = modbus.check_request(packet)
assert(rcode, "CHECK_NOT_OK")
assert(reply.get().data[1] == excode, "EXCODE_MISMATCH")
end
-- test the current packet, expecting success
function public.test_success__handle_packet()
local rcode, reply = modbus.handle_packet(packet)
assert(rcode, "CHECK_NOT_OK")
println(testutils.stringify(reply.get().data))
end
return public
end
return testutils