#!/opt/lnxall_app/bin/lua local bit32 = require 'bit32' local posix = require 'posix' local cjson = require 'cjson' local invoker = require 'invoker' local gpio_output = { [1] = 479, --LSD [2] = 480, [3] = 481, --HSD [4] = 482, [5] = 483, [6] = 484, [7] = 485, --DO [8] = 486, [9] = 487, [10] = 488, [11] = 489, [12] = 490, [13] = 491, [14] = 492, } local gpio_input = { [1] = 39, --board [2] = 40, [3] = 41, [4] = 42, [5] = 495, --external [6] = 496, [7] = 497, [8] = 498, [9] = 499, [10] = 500, [11] = 501, [12] = 502, [13] = 503, [14] = 504, [15] = 505, [16] = 506, [17] = 507, [18] = 508, [19] = 509, [20] = 510, } local g_tty = { [1] = '/dev/ttyS4', [2] = '/dev/ttyS5', [3] = '/dev/ttyS7', [4] = '/dev/ttyS3', [5] = '/dev/ttyCH343USB0', [6] = '/dev/ttyCH343USB1', [7] = '/dev/ttyCH343USB2', [8] = '/dev/ttyCH343USB3', } local lan2selfip = '192.168.7.101' local lan2pingip = '192.168.7.100' local function netdev_hasipv4(ndev) local okay, output = invoker.invoke(invoker.OUTPUT, "ip", "addr", "show", "dev", ndev) if okay ~= 0 or type(output) ~= "string" then return false end local ipaddr = string.match(output, "inet%s+([%d%.]+)/") if not ipaddr then io.stderr:write(string.format("Error, no ipv4 found for '%s'\n", ndev)) io.stderr:flush() return false end io.stdout:write(string.format("IPv4 address found for '%s': %s\n", ndev, ipaddr)) io.stdout:flush() return true end local function writefile(fpath, fdata) local fh = io.open(fpath, "wb") if fh then local okay = fh:write(fdata) fh:close(); fh = nil if okay then return true end io.stderr:write(string.format("Error, failed to write file: %s\n", fpath)) else io.stderr:write(string.format("Error, failed to open file: %s\n", fpath)) end io.stderr:flush() return false end local function setup_gpio() for _, gpio_in in ipairs(gpio_input) do local path = string.format("/sys/class/gpio/gpio%s", gpio_in) if not posix.access(path) then writefile("/sys/class/gpio/export", string.format("%d\n", gpio_in)) end writefile(string.format("/sys/class/gpio/gpio%d/direction", gpio_in), "in\n") end for _, gpio_out in ipairs(gpio_output) do local path = string.format("/sys/class/gpio/gpio%s", gpio_out) if not posix.access(path) then writefile("/sys/class/gpio/export", string.format("%d\n", gpio_out)) end writefile(string.format("/sys/class/gpio/gpio%d/direction", gpio_out), "out\n") end end local function test_dido() local rval = true local err = {} setup_gpio() -- all set 1 for _, gpio_out in ipairs(gpio_output) do writefile(string.format("/sys/class/gpio/gpio%s/value", gpio_out),"1\n") end invoker.msleep(200) for i, gpio_in in ipairs(gpio_input) do local fpath = string.format("/sys/class/gpio/gpio%s/value", gpio_in) local output = invoker.readfile(fpath) if not string.match(output ,"0") then local tag = false rval = false err[#err + 1] = i if tag then err[#err + 1] = i end io.stdout:write(string.format("set 1, get DI%d = %s\n", i, output)) io.stdout:write(string.format("Error,DI%d\n", i)) io.stdout:flush() end end -- all set 0 for _, gpio_out in ipairs(gpio_output) do writefile(string.format("/sys/class/gpio/gpio%s/value", gpio_out),"0\n") end invoker.msleep(200) for i, gpio_in in ipairs(gpio_input) do local fpath = string.format("/sys/class/gpio/gpio%s/value", gpio_in) local output = invoker.readfile(fpath) if not string.match(output ,"1") then local tag = true rval = false for j=1,#err do if err[j] == i then tag = false end end if tag then err[#err + 1] = i end io.stdout:write(string.format("set 0, get DI%d = %s\n", i, output)) io.stdout:write(string.format("Error,DI%d\n", i)) io.stdout:flush() end end local final = "" if rval then io.stdout:write("Test DI/DO OKAY!!!\n") io.stdout:flush() elseif #err == 20 then final = "all" io.stdout:write("DIDO all error!!!\n") io.stdout:flush() return false, final else for j=1,#err do if j == 1 then final = err[j] else final = final .. "," .. err[j] end end io.stdout:write(string.format("%s is error\n", final)) io.stdout:flush() return false, final end return true, nil end local function test_rs485(p0, p1) local wtf = posix.stat(p0) if type(wtf) ~= "table" or wtf['type'] ~= 'character device' then io.stderr:write(string.format("Error, invalid RS485 device: %s\n", p0)) io.stderr:flush() return false end wtf = posix.stat(p1) if type(wtf) ~= "table" or wtf['type'] ~= 'character device' then io.stderr:write(string.format("Error, invalid RS485 device: %s\n", p1)) io.stderr:flush() return false end -- for debian based system, we have a very strong `stty invoker.invoke(0, 'stty', '-F', p0, "raw", "9600", "cs8", "-hupcl", "-onlcr", "-iexten", "-echo", "-echoe", "-echok", "-echoctl", "-echoke", "-parenb") invoker.invoke(0, 'stty', '-F', p1, "raw", "9600", "cs8", "-hupcl", "-onlcr", "-iexten", "-echo", "-echoe", "-echok", "-echoctl", "-echoke", "-parenb") local rawstr = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ' local rstr = invoker.randstr(0x8, rawstr) local tmpfile = "/tmp/tmp-" .. rstr local catpid = invoker.invoke(invoker.NOWAIT, "/bin/sh", "-c", string.format("TTY_READ=1 exec tty_rw %s > %s", p0, tmpfile)) if type(catpid) ~= "number" then io.stderr:write(string.format("Error, failed to execute 'cat %s'\n", p0)) io.stderr:flush() posix.unlink(tmpfile) return false end invoker.msleep(10) invoker.invoke(0, 'tty_rw', p1, string.format('=%s=%s=\n', rstr, rstr)) invoker.msleep(50) invoker.kill(catpid, posix.SIGKILL) invoker.kill(catpid, posix.SIGTERM) invoker.waitpid(catpid, false) local output = invoker.readfile(tmpfile, invoker.TRIMEND) posix.unlink(tmpfile) if type(output) ~= "string" or not string.find(output, rstr, 1, true) then io.stderr:write(string.format("Error, test RS485 port has failed: %s => %s\n", p1, p0)) if type(output) == "string" then io.stderr:write(string.format("\tExpected: %s, output: %s\n", rstr, output)) end io.stderr:flush() return false end io.stdout:write(string.format("INFO: test RS485 port OKAY: %s => %s\n", p1, p0)) if type(output) == "string" then io.stdout:write(string.format("\tExpected: %s, output: %s\n", rstr, output)) end io.stdout:flush() return true end local function setup_can(candev) invoker.invoke(0, 'ip', 'link', 'set', candev, 'down') invoker.msleep(100) invoker.invoke(0, 'ip', 'link', 'set', candev, 'type', 'can', 'bitrate', '25000') invoker.msleep(100) invoker.invoke(0, 'ip', 'link', 'set', candev, 'up') invoker.msleep(100) end local function can_test(p0, p1) setup_can(p0) setup_can(p1) local can_id = "123" local pattern = invoker.randstr(16,"12345678ABCDEF") local cdata = can_id .. "#" .. pattern local tmpfile = "/tmp/candump.txt" posix.unlink(tmpfile) local cpid = invoker.invoke(invoker.NOWAIT, "/bin/sh", "-c", string.format("candump -L -T 4000 %s > %s 2>/dev/null", p0, tmpfile)) invoker.msleep(50) invoker.invoke(0, "cansend", p1, cdata) invoker.msleep(500) invoker.waitpid(cpid, false) local output = invoker.readfile(tmpfile) posix.unlink(tmpfile) if type(output) ~= "string" or not string.find(output, cdata, 1, true) then io.stderr:write(string.format("Error,output=%s,cdata=%s\n",output,cdata)) io.stderr:write(string.format("Error, CAN test has failed: %s => %s\n", p1, p0)) io.stderr:flush() return false end io.stdout:write(string.format("CAN test OKAY: %s => %s\n", p1, p0)) io.stdout:flush() return true end local function update_resolv() local res = invoker.readfile('/etc/resolv.conf') local myres = "nameserver 114.114.114.114\nnameserver 8.8.8.8\n" if res ~= myres then io.stdout:write("Updating /etc/resolv.conf...\n") io.stdout:flush() local filp = io.open('/etc/resolv.conf', 'wb') if filp then filp:write(myres) filp:close() filp = nil end end return true end local function ping_with(ifname, host) if not host then host = 'mqtt.lnxall.com' end if not netdev_hasipv4(ifname) then return false end io.stdout:write(string.format("Ping '%s' with interface '%s'...\n", host, ifname)) io.stdout:flush() local cmds = { "ping", "-c10", "-I", ifname, host } local _, output = invoker.invoke(invoker.OUTPUT + invoker.NOSTDIO, cmds) if type(output) == "string" and string.find(output, "packet loss", 1, true) then if string.find(output, "100% packet loss", 1, true) then return false end return true end return false end local function get_lan_host() local ipaddr = {} local arpfile = io.open("/proc/net/arp", "rb") if not arpfile then io.stderr:write("Error, failed to open arp file.\n") io.stderr:flush() return ipaddr end for line in arpfile:lines() do if string.find(line, "br-lan", 1, true) then local ip = string.match(line, "^([%d%.]+)") if invoker.isipv4(ip) then ipaddr[#ipaddr + 1] = ip end end end if #ipaddr == 0 then io.stderr:write("Error, no device connected to LAN\n") io.stderr:flush() end return ipaddr end local function test_led() local result = {} local leds = { "PL0", "PL1", "PL2", "PL3", "PL4" } for idx = 1, 8 do local wval = string.format("%d\n", bit32.band(idx, 0x1)) for _, led in ipairs(leds) do local okay = false local ledp = string.format("/sys/class/leds/%s/brightness", led) if posix.access(ledp) == 0 then local lfile = io.open(ledp, "wb") if lfile then okay = true lfile:write(wval) lfile:close() lfile = nil end end if okay then result[led] = { interfaces = led, error_code = 0, error_msg = "success" } else result[led] = { interfaces = led, error_code = 1, error_msg = string.format("led_%s not found", led) } end end invoker.msleep(250) end local result_led = {} result_led[1] = result["PL0"] result_led[2] = result["PL1"] result_led[3] = result["PL2"] result_led[4] = result["PL3"] result_led[5] = result["PL4"] return result_led end local function write_random_file(filepath) local n = 1 local filename = filepath .. ".bin" while true do if not posix.stat(filename) then break end filename = filepath .. "_" .. n .. ".bin" n = n + 1 io.stdout:write("File exists, using filename: " .. filename .. "\n") io.stdout:flush() end local orig_md5_cmd = "dd if=/dev/urandom bs=8192 count=1 2>/dev/null | tee " .. filename .. " | md5sum - | awk '{print $1}'" local orig_md5 = io.popen(orig_md5_cmd):read("*a"):gsub("\n", "") os.execute("sync") os.execute("echo 3 > /proc/sys/vm/drop_caches") local file = io.open(filename, "rb") if not file then io.stdout:write("Failed to open file: " .. filename .. "\n") io.stdout:flush() return false end file:seek("end") local size = file:seek() file:close() if size ~= 8192 then io.stdout:write("File size is not 8192 bytes: " .. filename .. "\n") io.stdout:flush() return false end local written_md5_cmd = "md5sum " .. filename .. " | awk '{print $1}'" local written_md5 = io.popen(written_md5_cmd):read("*a"):gsub("\n", "") if orig_md5 ~= written_md5 then io.stdout:write("File content is not the same: " .. filename .. "\n") io.stdout:flush() return false end os.execute("rm -f " .. filename) io.stdout:write(string.format("%s write success.\n",filename)) io.stdout:flush() return true end local function test_ssd(level) if not invoker.mountpoint('/mnt/ssd') then io.stdout:write("SSD not mounted.\n") io.stdout:flush() return false end if not posix.stat('/dev/nvme0n1') or posix.stat('/dev/nvme0n1')['type'] ~= 'block device' then io.stdout:write("SSD not found.\n") io.stdout:flush() return false end local okay, output = invoker.invoke(invoker.OUTPUT, "nvme", "smart-log", "/dev/nvme0n1") if okay ~= 0 or type(output) ~= "string" then io.stdout:write("Failed to get SSD health data.\n") io.stdout:flush() return false end local value_spare = string.match(output, "available_spare%s*:%s*(%d+)") local value_used = string.match(output, "percentage_used%s*:%s*(%d+)") local spare_num = value_spare and tonumber(value_spare) or 0 local used_num = value_used and tonumber(value_used) or 0 if spare_num and used_num then io.stdout:write(string.format("SSD result - spare_num: %d%%, used_num: %d%%\n", spare_num, used_num)) io.stdout:flush() if level == "1" and spare_num >= 90 and used_num <= 10 then return true end if level == "0" and spare_num >= 70 and used_num <= 30 then return true end end io.stdout:write("Failed to get SSD health data or SSD not health.\n") io.stdout:flush() return false end local function test_mmc(level) local device = "" if posix.stat('/dev/mmcblk0') and posix.stat('/dev/mmcblk0')['type'] == 'block device' then device = "/dev/mmcblk0" elseif posix.stat('/dev/mmcblk1') and posix.stat('/dev/mmcblk1')['type'] == 'block device' then device = "/dev/mmcblk1" else io.stderr:write("Error, not found mmc device\n") io.stderr:flush() return false end local okay, output = invoker.invoke(invoker.OUTPUT, "mmc", "extcsd", "read", device) if okay ~= 0 or type(output) ~= "string" then io.stderr:write("Error, failed to read MMC extcsd\n") io.stderr:flush() return false end local value_A = string.match(output, "Life Time Estimation A.-:%s*0x([%x]+)") local value_B = string.match(output, "Life Time Estimation B.-:%s*0x([%x]+)") local dec_value_A = value_A and tonumber(value_A, 16) or 0 local dec_value_B = value_B and tonumber(value_B, 16) or 0 local percent_A = dec_value_A * 10 local percent_B = dec_value_B * 10 if percent_A and percent_B then io.stdout:write(string.format("MMC result - percent_A: %d%%, percent_B: %d%%\n", percent_A, percent_B)) io.stdout:flush() if level == "1" and percent_A <= 10 and percent_B <= 10 then return true end if level == "0" and percent_A <= 30 and percent_B <= 30 then return true end end io.stdout:write("Failed to read MMC extcsd or MMC not health.\n") io.stdout:flush() return false end local function maindiag(arg1,arg2) finalsummary = {} local summary = {} local okay, facini = invoker.invoke(invoker.OUTPUT, "factory", "get") if okay ~= 0 or type(facini) ~= "string" then io.stderr:write("Error, 'factory get' has failed.\n") io.stderr:flush() return 1 end if not arg2 then io.stdout:write("Error, invaild input, set mmc/ssd lever to 10%.\n") io.stdout:flush() arg2 = "1" end local gwsn = string.match(facini, "SN=([^\r\n]+)") local hwver = string.match(facini, "HWVER=([^\r\n]+)") or "" local macaddr = string.match(facini, "MAC=([^\r\n]+)") or "" local gwpro = string.match(facini, "PRODUCT=([^\r\n]+)") if not gwsn or not gwpro then io.stderr:write("Error, gateway SN or PRODUCT not found.\n") io.stderr:flush() return 2 end local softver = invoker.readfile("/lib/dyiot/issue", invoker.TRIMEND) if type(softver) ~= "string" or #softver == 0 then softver = invoker.readfile("/etc/issue", invoker.TRIMEND) if type(softver) ~= "string" or #softver == 0 then softver = "unknown" end end invoker.invoke(0, "ifconfig", "br-lan:0", lan2selfip, "netmask", "255.255.255.0", "up") finalsummary["sn"] = gwsn finalsummary["timestamp"] = os.time() finalsummary["MAC"] = macaddr finalsummary["HWVER"] = hwver finalsummary["SWVER"] = softver summary["ioDetect"] = test_led() local ports = {} if test_rs485(g_tty[1], g_tty[2]) then ports[#ports + 1] = { interfaces = "RS485_1", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_1", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[2], g_tty[1]) then ports[#ports + 1] = { interfaces = "RS485_2", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_2", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[3], g_tty[4]) then ports[#ports + 1] = { interfaces = "RS485_3", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_3", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[4], g_tty[3]) then ports[#ports + 1] = { interfaces = "RS485_4", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_4", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[5], g_tty[6]) then ports[#ports + 1] = { interfaces = "RS485_5", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_5", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[6], g_tty[5]) then ports[#ports + 1] = { interfaces = "RS485_6", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_6", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[7], g_tty[8]) then ports[#ports + 1] = { interfaces = "RS485_7", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_7", error_msg = "CANNOT WORK", error_code = 1 } end if test_rs485(g_tty[8], g_tty[7]) then ports[#ports + 1] = { interfaces = "RS485_8", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RS485_8", error_msg = "CANNOT WORK", error_code = 1 } end if can_test('lnxcan1', 'lnxcan2') or can_test('lnxcan1', 'lnxcan2') or can_test('lnxcan1', 'lnxcan2') or can_test('lnxcan1', 'lnxcan2') or can_test('lnxcan1', 'lnxcan2')then ports[#ports + 1] = { interfaces = "CAN0", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "CAN0", error_msg = "CANNOT WORK", error_code = 1 } end if can_test('lnxcan2', 'lnxcan1') or can_test('lnxcan2', 'lnxcan1') or can_test('lnxcan2', 'lnxcan1') or can_test('lnxcan2', 'lnxcan1') or can_test('lnxcan2', 'lnxcan1')then ports[#ports + 1] = { interfaces = "CAN1", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "CAN1", error_msg = "CANNOT WORK", error_code = 1 } end if posix.access('/dev/rtc0') == 0 then ports[#ports + 1] = { interfaces = "RTC", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "RTC", error_msg = "RTC0 NOT FOUND", error_code = 1 } end local rval, binary = test_dido() if rval then ports[#ports + 1] = { interfaces = "DI_TEST", error_code = 0, error_msg = "success" } ports[#ports + 1] = { interfaces = "DO_TEST", error_code = 0, error_msg = "success" } else ports[#ports + 1] = { interfaces = "DI_TEST", error_code = 1, error_msg = binary } ports[#ports + 1] = { interfaces = "DO_TEST", error_code = 1, error_msg = binary } end if test_ssd(arg2) and write_random_file("/mnt/ssd/random_data") then ports[#ports + 1] = { interfaces = "SSD", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "SSD", error_msg = "Solid State Drive Error", error_code = 1 } end if test_mmc(arg2) and write_random_file("/random_data") then ports[#ports + 1] = { interfaces = "MMC", error_msg = "success", error_code = 0 } else ports[#ports + 1] = { interfaces = "MMC", error_msg = "MMC NOT HEALTH", error_code = 1 } end summary["OnBoardPort"] = ports local netw = {} update_resolv() if ping_with('eth0') or ping_with('eth0', '114.114.114.114') then netw[#netw + 1] = { interfaces = "WAN", error_msg = "success", error_code = 0 } else netw[#netw + 1] = { interfaces = "WAN", error_msg = "NOT OKAY", error_code = 1 } end invoker.invoke(0,'nmcli', 'con', 'down' ,'wan') update_resolv() if ping_with('eth1') or ping_with('eth1', '114.114.114.114') then netw[#netw + 1] = { interfaces = "WAN2", error_msg = "success", error_code = 0 } else netw[#netw + 1] = { interfaces = "WAN2", error_msg = "NOT OKAY", error_code = 1 } end invoker.invoke(0,'nmcli', 'con', 'up' ,'wan') local landevs, lanokay = get_lan_host(), false for _, ip in pairs(landevs) do if ping_with('br-lan', ip) then lanokay = true break end end if lanokay then netw[#netw + 1] = { interfaces = "LAN", error_msg = "success", error_code = 0 } else netw[#netw + 1] = { interfaces = "LAN", error_msg = "NOT OKAY", error_code = 1 } end if ping_with('br-lan', lan2pingip) then netw[#netw + 1] = { interfaces = "LAN2", error_msg = "success", error_code = 0 } else netw[#netw + 1] = { interfaces = "LAN2", error_msg = "NOT OKAY", error_code = 1 } end summary["NetWork"] = netw finalsummary["data"] = summary io.stdout:write("\n---TEST SUMMARY---\ntest_result:") io.stdout:write(cjson.encode(finalsummary)) io.stdout:write("\n") io.stdout:flush() invoker.invoke(0, "ifconfig", "br-lan:0", lan2selfip, "netmask", "255.255.255.0", "down") end posix.setenv("PATH", "/opt/lnxall_app/bin:/usr/bin:/usr/sbin:/bin:/sbin", true) maindiag(arg[1],arg[2]) os.exit(0)