require "log" require "sys" require "utils" require "patch" require "pack" siganl = require 'posix.signal' syswait = require 'posix.sys.wait' unistd = require 'posix.unistd' mqtt=require "mosquitto" base64=require("base64") dyutils=require("dyutils") lpack=require("lua_pack") require "internal_api" require"lsqlite3" -- require"nodes_sta" RACK_PATH = "/app/" PROCJET = "SDB_" THRESHOLD_FILE = RACK_PATH .. PROCJET .. "threshold.json" SCHEDULE_FILE = RACK_PATH .. PROCJET .. "schedule.json" LEAKCURREBT_FILE = RACK_PATH .. PROCJET .. "leakcurrent.txt" ALARM_CONFLICT = 6 ALARM_OFFLINE = 5 ALARM_SENSOR_SC = 4 ALARM_SENSOR_OD = 3 ALARM_UNDERMIN = 2 ALARM_OVERMAX = 1 SWITCH_MANUAL = 0 SWITCH_AUTO = 1 TIMEZOEN = 8*3600 ---- 秒 VALIDPERIOD = 7 * 24 *3600 ---- 秒 REPORT_PERIOD = 60*60 ---- 秒 ZENITH = 90.833 PI = 3.1415926535898 MYSELF_SN_SUFFIX = "_RTU" -- 风扇开启阈值 THRESHOLD_OPENFAN=60 TEST_LEAK_CURRENT_HOUR = 8 -- 互感器范围 TRANSFORMER_RANGE=50 -- 各项合计最大电流 MAX_LIMIT_CURRENT=10 local realData = {} local realAlarm = {} local HistoryData = {} local historyAlarm = {} local scheduleCircuits={Circuit1={},Circuit2={},Circuit3={},Circuit4={}} local cacheAlarm = {} local sampleTime = {} local controlCache = {} local using_job = {} local start_time = {} local end_time = {} local manu_cmd = {} local conflictFlagTime = {} local last_report = 0 local ScheOffset = 10 ---- 分 local Longitude,Latitude = 120.12,30.16 local protactflags = { overCurrent = false, } local alarmMap = { ---- |报警ID名称 | 需要源数据的字段 | | 解析后设备字段 | 用源数据 | 用逻辑数据 | ---- 源设备SN, 逻辑设备 源服务key 一级对象 二级对象 timeout min max low high {"Envir.Temp", "_ENV", "shujushangbao1", "temperature", "Envir", "Temp", 300, nil, nil, -20, 60 }, {"Envir.Humidity", "_ENV", "shujushangbao1", "humidity", "Envir", "Humidity", 300, nil, nil, 0, 80 }, {"Meter", "_METER", "post", "APhaseVoltage", "Meter", "Ua", 60, nil, nil, 200, 260 }, {"LeakCurr.AnodeRes", "_AI", "LeakageCurrent", "AnodeRes", "LeakageCurrent", "AnodeRes", 60, nil, nil, 200, nil }, {"LeakCurr.CathodeRes", "_AI", "LeakageCurrent", "CathodeRes", "LeakageCurrent", "CathodeRes", 60, nil, nil, 200, nil }, {"Circuit1.Current", "_AI", "ReadData", "aag1", "Circuit1", "Current", 60, 2000, 20000, 0, MAX_LIMIT_CURRENT/2 }, {"Circuit2.Current", "_AI", "ReadData", "aag2", "Circuit2", "Current", 60, 2000, 20000, 0, MAX_LIMIT_CURRENT/2 }, {"Circuit3.Current", "_AI", "ReadData", "aag3", "Circuit3", "Current", 60, 2000, 20000, 0, MAX_LIMIT_CURRENT/2 }, {"Circuit4.Current", "_AI", "ReadData", "aag4", "Circuit4", "Current", 60, 2000, 20000, 0, MAX_LIMIT_CURRENT/2 }, -- {"Circuit1.Voltage", "_AI", "Read_DCvoltage", "aag6", "Circuit1", "Voltage", 60, 2000, 20000, 180, 250 }, -- {"Circuit2.Voltage", "_AI", "Read_DCvoltage", "aag6", "Circuit2", "Voltage", 60, 2000, 20000, 180, 250 }, -- {"Circuit3.Voltage", "_AI", "Read_DCvoltage", "aag6", "Circuit3", "Voltage", 60, 2000, 20000, 180, 250 }, -- {"Circuit4.Voltage", "_AI", "Read_DCvoltage", "aag6", "Circuit4", "Voltage", 60, 2000, 20000, 180, 250 }, -- 下面两组参数的比较值不在实时数据里 {"OverCurrent", nil, nil, nil, nil, nil, nil, nil, nil, nil, MAX_LIMIT_CURRENT*0.9 }, {"CurrentProtact", nil, nil, nil, nil, nil, nil, nil, nil, nil, MAX_LIMIT_CURRENT*1.1 }, {"Circuit1.Feedback","_DIO", "RData", "DI0", nil, nil, 60, nil, nil, nil, nil}, {"Circuit2.Feedback","_DIO", "RData", "DI1", nil, nil, 60, nil, nil, nil, nil}, {"Circuit3.Feedback","_DIO", "RData", "DI2", nil, nil, 60, nil, nil, nil, nil}, {"Circuit4.Feedback","_DIO", "RData", "DI3", nil, nil, 60, nil, nil, nil, nil}, {"CircuitMain.Feedback","_DIO","RData", "DI11", nil, nil, 60, nil, nil, nil, nil}, {"Security.FrontDoor","_DIO", "RData", "DI4", "Security", "FrontDoor", 60, nil, nil, nil, 1}, {"Security.BackDoor", "_DIO", "RData", "DI5", "Security", "BackDoor", 60, nil, nil, nil, 1}, {"Security.ManualSW", "_DIO", "RData", "DI7", "Security", "ManualSW", 60, nil, nil, nil, 1}, {"Security.LR", "_DIO", "RData", "DI8", "Security", "LR", 60, nil, nil, nil, 1}, {"Security.FloatBall","_DIO", "RData", "DI9", "Security", "FloatBall", 60, nil, nil, nil, 1}, {"Security.WaterSoaked","_DIO","RData", "DI10", "Security", "WaterSoaked", 60, nil, nil, nil, 1}, } local transformer_k = TRANSFORMER_RANGE and TRANSFORMER_RANGE/16000 or 0.003125 local transformer_b = TRANSFORMER_RANGE and TRANSFORMER_RANGE/-4 or -12.5 local dataCollectMap = { ---- | 源设备 | | 逻辑设备 | 数值变化监控 | ---- SN后缀, 源设备 源服务key 一级对象 二级对象 偏移 系数 变化值 {"_ENV", "shujushangbao1", "temperature", "Envir", "Temp", 0, 1, 1}, {"_ENV", "shujushangbao1", "humidity", "Envir", "Humidity", 0, 1, 2}, {"_AI", "ReadData", "aag1", "Circuit1", "Current", transformer_b, transformer_k, 1}, {"_AI", "ReadData", "aag2", "Circuit2", "Current", transformer_b, transformer_k, 1}, {"_AI", "ReadData", "aag3", "Circuit3", "Current", transformer_b, transformer_k, 1}, {"_AI", "ReadData", "aag4", "Circuit4", "Current", transformer_b, transformer_k, 1}, {"_AI", "Read_DCvoltage", "aag6", "Circuit1", "Voltage", -125, 0.03125, 10}, {"_AI", "Read_DCvoltage", "aag6", "Circuit2", "Voltage", -125, 0.03125, 10}, {"_AI", "Read_DCvoltage", "aag6", "Circuit3", "Voltage", -125, 0.03125, 10}, {"_AI", "Read_DCvoltage", "aag6", "Circuit4", "Voltage", -125, 0.03125, 10}, {"_AI", "LeakageCurrent", "Anode", "LeakageCurrent", "Anode", -(0.25 + 50), 0.0000625, 1}, {"_AI", "LeakageCurrent", "Cathode", "LeakageCurrent", "Cathode", -(0.25 + 50), 0.0000625, 1}, {"_AI", "LeakageCurrent", "AnodeRes", "LeakageCurrent", "AnodeRes", 0, 1, 100}, {"_AI", "LeakageCurrent", "CathodeRes", "LeakageCurrent", "CathodeRes", 0, 1, 100}, {"_DIO", "RData", "DI0", "Circuit1", "Feedback", 0, 1, 1}, {"_DIO", "RData", "DI1", "Circuit2", "Feedback", 0, 1, 1}, {"_DIO", "RData", "DI2", "Circuit3", "Feedback", 0, 1, 1}, {"_DIO", "RData", "DI3", "Circuit4", "Feedback", 0, 1, 1}, {"_DIO", "RData", "DI11", "CircuitMain", "Feedback", 0, 1, 1}, {"_DIO", "RData", "DI4", "Security", "FrontDoor", 0, 1, 1}, {"_DIO", "RData", "DI5", "Security", "BackDoor", 0, 1, 1}, {"_DIO", "RData", "DI7", "Security", "ManualSW", 0, 1, 1}, {"_DIO", "RData", "DI8", "Security", "LR", 0, 1, 1}, {"_DIO", "RData", "DI9", "Security", "FloatBall", 0, 1, 1}, {"_DIO", "RData", "DI10", "Security", "WaterSoaked", 0, 1, 1}, {"_METER", "post", "APhaseVoltage", "Meter", "Ua", 0, 1, 10}, {"_METER", "post", "BPhaseVoltage", "Meter", "Ub", 0, 1, 10}, {"_METER", "post", "CPhaseVoltage", "Meter", "Uc", 0, 1, 10}, {"_METER", "post", "APhaseCurrent", "Meter", "Ia", 0, 1, 0.5}, {"_METER", "post", "BPhaseCurrent", "Meter", "Ib", 0, 1, 0.5}, {"_METER", "post", "CPhaseCurrent", "Meter", "Ic", 0, 1, 0.5}, {"_METER", "post", "TotalPowerConsumption", "Meter", "Energy", 0, 1, 5}, {"_METER", "post", "totalForwardElectricEnergy", "Meter", "PositiveEnergy", 0, 1, 5}, {"_METER", "post", "totalReverseElectricEnergy", "Meter", "ReverseEnergy", 0, 1, 5}, } local controlMap = { ---- | 源设备 | | 逻辑设备 | ---- SN后缀, 源设备 源服务key 对象 {"_DIO", "DO0", "open", "Circuit1" }, {"_DIO", "DO1", "open", "Circuit2" }, {"_DIO", "DO2", "open", "Circuit3" }, {"_DIO", "DO3", "open", "Circuit4" }, -- {"_DO", "DO4", "open", "SwAnode" }, -- {"_DO", "DO5", "open", "SwCathode" }, {"_DIO", "DO6", "open", "Fan" }, {"_DIO", "DO8", "open", "CircuitMain" }, } local function calRiseSet(date, isRise) local lngHour = Longitude / 15; local t = date + (isRise and 6.0 or 18.0 - lngHour) / 24 local M = 0.9856 * t - 3.289; local L = M + 1.916 * math.sin(M * PI / 180) + 0.02 * math.sin(2 * M * PI / 180) + 282.634 while true do if L >= 360 then L = L- 360 else if L < 0 then L = L + 360 else break end end end local RA = math.atan(0.91764 * math.tan(L * PI / 180)) / PI * 180 while true do if RA >= 360 then RA = RA - 360 else if RA < 0 then RA = RA + 360 else break end end end local Lquadrant = (math.floor(L /90)) * 90 local RAquadrant = (math.floor(RA /90)) * 90 RA = RA + Lquadrant - RAquadrant RA = RA / 15 local sinDec = 0.39782 * math.sin(L * PI / 180) local cosDec = math.cos(math.asin(sinDec)) local cosH = (math.cos(ZENITH * PI / 180) - sinDec * math.sin(Latitude * PI / 180)) / (cosDec * math.cos(Latitude * PI / 180)) local H = 0 if isRise then H = 360.0 - math.acos(cosH) / PI * 180 else H = math.acos(cosH) / PI * 180 end H = H / 15; local T = H + RA - 0.06571 * t - 6.622; while true do if T >= 24 then T = T - 24 else if T < 0 then T = T + 24 else break end end end return T * 60 + (isRise and ScheOffset or - ScheOffset) end local function paser_special(obj,tags) local src_sn_suffix = "_AI" local src_service = "Read_DCleakagecurrent" local src_key = "aag5" if src_sn_suffix == string.sub(obj.sn,#obj.sn - #src_sn_suffix + 1) and src_service == obj.identifier and tags[src_key] then local value = tags[src_key] if value and type(value) == "number" then sys.publish("AI4_SIGNAL",value) end end end local function realDataCala(obj,tags) for i,item in ipairs(dataCollectMap) do local src_sn_suffix = item[1] local src_service = item[2] local src_key = item[3] local des_l1_key = item[4] local des_l2_key = item[5] local offset = item[6] and item[6] or 0 local scale = item[7] and item[7] or 1 if src_sn_suffix == string.sub(obj.sn,#obj.sn - #src_sn_suffix + 1) and src_service == obj.identifier and tags[src_key] then if not realData[des_l1_key] then realData[des_l1_key] = {} end local tmp = realData[des_l1_key] if type(tags[src_key]) == "table" then log.warn("realDataCala", string.format("Key %s Need a numeric value",src_key)) return end local value = tags[src_key] * scale + offset tmp[des_l2_key] = (value < 0) and 0 or value end end end local function getTotalCurrent() local total_current = 0 for _,circuit in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do local data = realData[circuit] if data and data.Current then total_current = total_current + data.Current end end return total_current end local function realAlarmCala(obj,tags) local totalCurrent = getTotalCurrent() for i,item in ipairs(alarmMap) do local alarm_id = item[1] local src_sn_suffix = item[2] local src_service = item[3] local src_key = item[4] local des_l1_key = item[5] local des_l2_key = item[6] local timeout = item[7] local min = item[8] local max = item[9] local low = item[10] local high = item[11] local holdtime = item[12] local currTime = os.time() if not cacheAlarm[alarm_id] then cacheAlarm[alarm_id] = {false,false,false,false,false,false} end local tmpAlarm = cacheAlarm[alarm_id] local function updateAlarmItem(enum,condition) -- log.info("tag", enum,condition,enable) tmpAlarm[enum] = condition ~= nil and condition or false end if src_key and src_sn_suffix == string.sub(obj.sn,#obj.sn - #src_sn_suffix + 1) and src_service == obj.identifier and tags[src_key] then sampleTime[alarm_id] = currTime updateAlarmItem(ALARM_SENSOR_SC,min and tags[src_key] and tags[src_key] < min) ---- 计算短路和断线报警 updateAlarmItem(ALARM_SENSOR_OD,max and tags[src_key] and tags[src_key] >= max) end if des_l1_key and des_l2_key and realData[des_l1_key] then local tmpData = realData[des_l1_key] ---- 计算过上限和过下限报警 updateAlarmItem(ALARM_UNDERMIN,low and tmpData[des_l2_key] and tmpData[des_l2_key] < low) updateAlarmItem(ALARM_OVERMAX,high and tmpData[des_l2_key] and tmpData[des_l2_key] >= high) end if timeout then updateAlarmItem(ALARM_OFFLINE,timeout and sampleTime[alarm_id] and currTime > (sampleTime[alarm_id] + timeout)) end -- 比较总电流 if (alarm_id == "OverCurrent" or alarm_id == "CurrentProtact") and totalCurrent then -- 计算总电流过上限和过下限报警 updateAlarmItem(ALARM_UNDERMIN,low and totalCurrent < low) updateAlarmItem(ALARM_OVERMAX,high and totalCurrent > high) end local errorCode = 0 for index = ALARM_CONFLICT,ALARM_OVERMAX,-1 do -- 转换到报警code if tmpAlarm[index] then errorCode = index break end end realAlarm[alarm_id] = errorCode end if totalCurrent then if totalCurrent > MAX_LIMIT_CURRENT*1.1 then -- 触发保护 if not protactflags.overCurrent then log.info("protact",string.format("Over total current of system,TotalCurrent:%s limit:%s",totalCurrent,MAX_LIMIT_CURRENT)) protactflags.overCurrent = true end end end end local function writeDO(key,value) for _,item in ipairs(controlMap) do local src_sn_suffix = item[1] local src_service = item[2] local src_key = item[3] local des_key = item[4] if des_key == key then local obj = {} obj[src_key] = value local dev_sn,port,identifier,param = string.format("%s%s",gw_sn,src_sn_suffix),"RS485_3",src_service,obj internal_api.call_dev_service(gw_sn,dev_sn,port,identifier,param) end end end local function getSubDevice(gw_sn) local devs = {} for _,item in ipairs(dataCollectMap) do devs[string.format("%s%s",gw_sn,item[1])] = true end local devsarray = {} for key,flag in pairs(devs) do if flag then table.insert(devsarray,key) end end table.insert(devsarray,string.format("%s%s",gw_sn,MYSELF_SN_SUFFIX)) return devsarray end --[[local function controlDev(key,value) local found for _,item in ipairs(controlMap) do if item[4] == key then local dev_sn,port,param = string.format("%s%s",gw_sn,item[1]),"RS485_3" param[item[3] ] = value internal_api.call_dev_service(gw_sn,dev_sn,port,item[2],param) end end if not found then log.warn("controlDev", "Invalid cmd",key, value) end end]] local function get_leak_current() local obj = {} local dio_sn = string.format("%s_DIO",gw_sn) -- 检查适合否和测试条件 for _,circuit in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do local data = realData[circuit] if data and data.Feedback and data.Feedback == 1 then log.error("leak_current", string.format("request was refused! Need turn off %s before testing",circuit) ) return end end local ai_sn,port,ai_identifier = string.format("%s_AI",gw_sn),"RS485_3","Read_DCleakagecurrent" internal_api.call_dev_service(gw_sn,dio_sn,port,"DO8",{open=1}) internal_api.call_dev_service(gw_sn,dio_sn,port,"DO4",{open=0}) internal_api.call_dev_service(gw_sn,dio_sn,port,"DO5",{open=1}) sys.wait(500) internal_api.call_dev_service(gw_sn,ai_sn,port,ai_identifier) local res,data = sys.waitUntil("AI4_SIGNAL", 5000) if not res then log.warn("leak_current", "Timeout by wait AI") else realDataCala({sn=ai_sn,identifier="LeakageCurrent"},{Cathode=data}) end internal_api.call_dev_service(gw_sn,dio_sn,port,"DO4",{open=1}) sys.wait(500) internal_api.call_dev_service(gw_sn,ai_sn,port,ai_identifier) local res,data = sys.waitUntil("AI4_SIGNAL", 5000) if not res then log.warn("leak_current", "Timeout by wait AI") else realDataCala({sn=ai_sn,identifier="LeakageCurrent"},{Anode=data}) end internal_api.call_dev_service(gw_sn,dio_sn,port,"DO4",{open=0}) internal_api.call_dev_service(gw_sn,dio_sn,port,"DO5",{open=0}) internal_api.call_dev_service(gw_sn,dio_sn,port,"DO8",{open=0}) local voltage local tags = realData["Circuit1"] if not voltage and tags and tags.Voltage then voltage = tags.Voltage end local tags = realData["LeakageCurrent"] if tags then obj.Cathode = tags.Anode obj.CathodeRes = voltage/tags.Cathode realDataCala({sn=ai_sn,identifier="LeakageCurrent"},{CathodeRes=obj.CathodeRes}) end if tags then obj.Anode = tags.Anode obj.AnodeRes = voltage/tags.Anode realDataCala({sn=ai_sn,identifier="LeakageCurrent"},{AnodeRes=obj.AnodeRes}) end return obj end local function strDate2Tab(str) local dateTab = {} local s1,s2,s3,s4,s5,s6 = string.match(str,'(%d+)-(%d+)-(%d+) (%d+):(%d+):(%d+)') if not s1 or not s2 or not s3 or not s4 or not s5 or not s6 then log.error("date","Invalid format:string",str,"example:2020-07-15 16:39:00") return end dateTab.year = tonumber(s1) dateTab.month = tonumber(s2) dateTab.day = tonumber(s3) dateTab.hour = tonumber(s4) dateTab.min = tonumber(s5) dateTab.sec = tonumber(s6) return dateTab end local function strDate2Time(str) local date_str = strDate2Tab(str) return os.time({year =date_str.year, month = date_str.month, day =date_str.day, hour = date_str.hour, min = date_str.min, sec = date_str.sec }) end local function noticeProcess(circuit,value,source) log.info("notice",string.format("%s %s Publish [%s] = %d",os.date("%Y-%m-%d %H:%M:%S",os.time() + TIMEZOEN),source and source or "",circuit,value)) controlCache[circuit] = value manu_cmd[circuit] = nil end local function diffTime(str) ---- 对比完整时间 local date = os.date("*t",os.time() + TIMEZOEN) ---- local time = strDate2Time(str) -- log.info("schedule",string.format( "Schedule time:%d",time)) local currTime = os.time({year =date.year, month = date.month, day =date.day, hour = date.hour, min = date.min, sec = date.sec }) -- log.info("schedule",string.format( "Current time:%d",currTime)) return time - currTime end local function diffDayTime(str) local date = os.date("*t",os.time() + TIMEZOEN) ---- 找到今天的策略 local date_str = strDate2Tab(str) local time = os.time({year =date_str.year, month = date_str.month, day =date_str.day, hour = 0, min = 0, sec = 0 }) -- log.info("schedule",string.format( "Schedule time:%d",time)) local currTime = os.time({year =date.year, month = date.month, day =date.day, hour = 0, min = 0, sec = 0 }) -- log.info("schedule",string.format( "Current time:%d",currTime)) return time - currTime end local function isTimeUp(str) local date = os.date("*t",os.time() + TIMEZOEN) ---- 当天时间到 local date_str = strDate2Tab(str) local time = os.time({year =date_str.year, month = date_str.month, day =date_str.day, hour = date_str.hour, min = date_str.min, sec = date_str.sec }) local currTime = os.time({year =date.year, month = date.month, day =date.day, hour = date.hour, min = date.min, sec = date.sec }) return currTime >= time end local function lookupSchedule(key) local schedule_index local schedule = nil local isNew local schedules = scheduleCircuits[key] if schedules and #schedules >= 1 then for index,sche in ipairs(schedules) do local diff = diffDayTime(sche.StartDate) if diff == 0 then ---- 同一天 isNew = true schedule = sche schedule_index = index break end end end return schedule,schedule_index,isNew end local function lookupDefaultJob(key) local schedule = {} local date = os.date("*t",os.time() + TIMEZOEN) ---- 当天时间到 local min = calRiseSet(date.yday, false) ---- 日落->开灯时间 schedule.StartDate = string.format("%04d-%02d-%02d %02d:%02d:00",date.year, date.month, date.day, math.floor(min / 60),math.floor(min % 60) ) local date = os.date("*t",os.time() + TIMEZOEN + 24*3600) ---- 次日关灯时间 local min = calRiseSet(date.yday, true) ---- 日出->关灯时间 schedule.EndDate = string.format( "%04d-%02d-%02d %02d:%02d:00",date.year, date.month,date.day, math.floor(min / 60),math.floor(min % 60) ) return schedule end local function default_job() for _,circuit in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do sys.taskInit(function(key) while(true) do sys.wait(100) if os.time() < 946656000 then sys.wait(60000) else local schedule while not using_job[key] do sys.wait(10000) schedule = lookupDefaultJob(key) log.info(key,"Time of latitude and longitude",schedule.StartDate,schedule.EndDate) if schedule then start_time[key] = schedule.StartDate end_time[key] = schedule.StartDate if isTimeUp(schedule.StartDate) then ---- 等待结束时间没过去,才判断开始时间 noticeProcess(key,1,"default") break end else start_time[key] = nil end_time[key] = nil end end while not using_job[key] do sys.wait(1000) if isTimeUp(schedule.EndDate) then ---- 等待结束 noticeProcess(key,0,"default") break end end end ---- 当前时间必须是2000年之后,才继续工作 end end,circuit) end end local function schedule_monitor() for _,circuit in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do sys.taskInit(function(key) while(true) do sys.wait(100) if os.time() < 946656000 then log.warn("system", "RTC lost!!!") sys.wait(60000) else local schedule_index local schedule local isNew while true do sys.wait(1000) schedule,schedule_index,isNew = lookupSchedule(key) using_job[key] = schedule and true or false if schedule then start_time[key] = schedule.StartDate end_time[key] = schedule.StartDate if isTimeUp(schedule.StartDate) then ---- 等待结束时间没过去,才判断开始时间 noticeProcess(key,1,"remote") break end end end while true do sys.wait(1000) if schedule then if not schedule.EndDate then log.warn("schedule","Schedule has been remove in running") break end if isTimeUp(schedule.EndDate) then ---- 等待结束 noticeProcess(key,0,"remote") if isNew then ---- 将策略删除并写入历史策略 table.remove( scheduleCircuits[key],schedule_index) io.writeFile(SCHEDULE_FILE,cjson.encode(scheduleCircuits),"w+") end break end end end end ---- 当前时间必须是2000年之后,才继续工作 end end,circuit) end end local function leak_current() local last while(true) do local ret,dev_sn,identifier = sys.waitUntil("GET_LEAK_CURRENT",3000) local date = os.date("*t",os.time() + TIMEZOEN) ---- if ret then local obj = get_leak_current() internal_api.service_response(gw_sn,dev_sn,"NULL",identifier ,obj or {}) elseif date.hour == TEST_LEAK_CURRENT_HOUR and date.min == 0 then if not last or date.day ~= last.day or date.month ~= last.month or date.year ~= last.year then last = date local obj = get_leak_current() if not obj then return end if not obj.Cathode and not obj.Anode and not obj.CathodeRes and not obj.AnodeRes then return end local content = string.format("%02d-%02d-%02d %d %f %f %f %f\n",date.year,date.month,date.day,os.time(),obj.Anode,obj.Cathode,obj.AnodeRes,obj.CathodeRes) io.writeFile(LEAKCURREBT_FILE, content, "a+") end end end end local function thresholdEffect(obj) if not obj then log.warn("threshold","Update config failed.Invalid params") return end for _,item in ipairs(alarmMap) do local alarm_id = item[1] local threshold = obj[alarm_id] if threshold and type(threshold) == "table" then item[10] = threshold[1] item[11] = threshold[2] log.info("update threshold config",string.format("alarm[%s]=low:%d;high:%d;",alarm_id,item[10],item[11])) end end end local function load_config() local obj = {} if io.exists(THRESHOLD_FILE) then local str = io.readFile(THRESHOLD_FILE) if str and #str > 0 then obj = cjson.decode(str or '') thresholdEffect(obj) end end if io.exists(SCHEDULE_FILE) then local str = io.readFile(SCHEDULE_FILE) if str and #str > 0 then obj = cjson.decode(str or '') scheduleCircuits = obj end end end local function iot2app(topic,payload) -- print("iot2app",topic,payload) if string.find(topic,MYSELF_SN_SUFFIX) then local obj = cjson.decode(payload) if obj.identifier == "SDBTotalState" then internal_api.service_response(gw_sn,obj.sn,"NULL",obj.identifier,realData,obj.mi) elseif obj.identifier == "SDBSetState" then if obj["Fan"] then writeDO("Fan",obj["Fan"]) end for _,key in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do local value = obj[key] if value and type(value) == "number" then controlCache[key] = value manu_cmd[key] = value log.info("control", string.format("Remote control[%s]=%d",key,value)) end end elseif obj.identifier == "SDBGetSchedules" then if obj.StartDate and obj.EndDate then if not obj.Filter then obj.Filter = {"Circuit1","Circuit2","Circuit3","Circuit4"} end local scheTagt = {} for _,key in ipairs(obj.Filter) do local schedules = scheduleCircuits[key] for index,schedule in ipairs(schedules) do if strDate2Time(obj.StartDate) <= strDate2Time(schedule.StartDate) and strDate2Time(schedule.EndDate) >= strDate2Time(obj.EndDate) then log.info("schedule","Avaiable schedule",string.format("End date:%s;Start date:%s",schedule.EndDate,schedule.StartDate)) if not scheTagt[key] then scheTagt[key] = {} end local sches = scheTagt[key] table.insert(sches, schedule) ---- NOTE:已经执行的策略必须走到结束,无法删除 if #sches >= 100 then log.warn("schedule","Select too more schedules,must be less than 50",key) break end end end end internal_api.service_response(gw_sn,obj.sn,"NULL",obj.identifier,scheTagt) else log.warn("control", "Invalid params",obj.StartDate,obj.EndDate) end elseif obj.identifier == "SDBSSetThreshold" then local thd = {} for _,item in ipairs(alarmMap) do local alarm_id = item[1] thd[alarm_id] = obj[alarm_id] end thresholdEffect(thd) io.writeFile(THRESHOLD_FILE,cjson.encode(thd),"w+") elseif obj.identifier == "SDBSSetSchedules" then sys.publish("SCHEDULE_CHANGED",obj) elseif obj.identifier == "SDBGetLeakCurrent" then sys.publish("GET_LEAK_CURRENT",obj.sn,obj.identifier) elseif obj.identifier == "SDBProtactRelieve" then if obj.overCurrent == 1 then protactflags.overCurrent = false end elseif obj.identifier == "SDBDoFactory" then os.remove(THRESHOLD_FILE) os.remove(SCHEDULE_FILE) os.remove(LEAKCURREBT_FILE) os.execute("/etc/init.d/protocol_parser restart") end end end local function circuit_enabled() local numbers = 0 for _,key in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do if not realData[key] then realData[key] = {} end local data = realData[key] if data and data.Ctrl == 1 then numbers = numbers + 1 end end return numbers end local function control_priority() local lastManualSW while(true) do sys.wait(1000) local str = os.date("%Y-%m-%d %H:%M:%S",os.time() + TIMEZOEN) local currDate = string.gsub(str,"(%w+) .*","%1" ) for _,key in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do if not realData[key] then realData[key] = {} end local data = realData[key] local value if protactflags.overCurrent then -- 过流保护联动逻辑 value = 0 else local security = realData["Security"] if security and security.ManualSW then -- 没用拿到手自动开关状态,不做任何动作 if lastManualSW ~= security.ManualSW then log.warn("control", "Manual switch status:",security.ManualSW) lastManualSW = security.ManualSW end if security.ManualSW == SWITCH_MANUAL then -- 直接写0 value = 0 elseif security.ManualSW == SWITCH_AUTO then --自动模式从控制cache拿值 value = controlCache[key] end end end if value~= nil then -- log.warn("control", string.format("Control[%s] value:%d",key,value)) if type(value) ~= "number" then log.warn("control", string.format("Control[%s] value must input type of number",key)) else if not data.Feedback or data.Feedback ~= value then local tran = realData["CircuitMain"] if not tran then log.info("control","Cmd is ignored as invaild object about [CircuitMain]") sys.wait(500) else if value == 0 then -- 如果是关闭回路,检查是否是最后一路 -- log.info("control",circuit_enabled(),value,tran.Feedback,tran.Ctrl) if circuit_enabled() == 1 then tran.Ctrl = 0 while(not tran.Feedback or tran.Feedback ~= tran.Ctrl) do log.info("control", string.format("[%s]Control=%d,Feedback=%d","CircuitMain",tran.Ctrl,tran.Feedback and tran.Feedback or -1)) writeDO("CircuitMain",tran.Ctrl) sys.wait(1000) end end elseif value == 1 then -- 如果是开灯检查是否是第一路 log.info("control",circuit_enabled(),value,tran.Feedback,tran.Ctrl) if circuit_enabled() == 0 then tran.Ctrl = 1 while(not tran.Feedback or tran.Feedback ~= tran.Ctrl) do log.info("control", string.format("[%s]Control=%d,Feedback=%d","CircuitMain",tran.Ctrl,tran.Feedback and tran.Feedback or -1)) writeDO("CircuitMain",tran.Ctrl) sys.wait(1000) end end end data.Ctrl = value log.info("control", string.format("[%s]Control=%d,Feedback=%d",key,data.Ctrl,data.Feedback and data.Feedback or -1)) writeDO(key,value) sys.wait(500) end elseif data.Feedback == value then --如果反馈 和设置值完全一样 data.Ctrl = value end end end end if THRESHOLD_OPENFAN then local data = realData["Envir"] if data and data.Temp and data.Temp > THRESHOLD_OPENFAN and (not controlCache["Fan"] or controlCache["Fan"] ~= 1) then controlCache["Fan"] = 1 writeDO("Fan",controlCache["Fan"]) log.info("control", "Fan is trun ON") elseif data and data.Temp and data.Temp < THRESHOLD_OPENFAN-1 and (not controlCache["Fan"] or controlCache["Fan"] ~= 0) then controlCache["Fan"] = 0 writeDO("Fan",controlCache["Fan"]) log.info("control", "Fan is trun OFF") end end end end local function schedule_merge() while(true) do local ret,obj = sys.waitUntil("SCHEDULE_CHANGED", 2000) if ret then local changed for _,key in ipairs({"Circuit1","Circuit2","Circuit3","Circuit4"}) do local schedules = obj[key] local useingSches = scheduleCircuits[key] while(true) do ---- 待生效 local strDate = nil local schedulelist = {} if not schedules or #schedules < 1 then break end -- log.info("tag", "schedules",#schedules,key,cjson.encode(schedules)) if #schedules > 100 then log.warn("schedule","Ignore schedules,must be less than 50",key) break end local index = 1 while(index <= #schedules) do schedule = schedules[index] if not schedule.StartDate or not schedule.EndDate then table.remove(schedules,index) log.warn("schedule","Removed schedule,Not fount start and end time") elseif diffTime(schedule.EndDate) < 0 then ---- 已经过去 table.remove(schedules,index) log.warn("schedule","Removed schedule,Stop time too later",string.format("End date:%s;Start date:%s",schedule.EndDate,schedule.StartDate)) elseif diffTime(schedule.StartDate) > 366 * 24 * 3600 then ---- 不接受1年以后的策略 table.remove(schedules,index) log.warn("schedule","Removed schedule,Date too early",string.format("End date:%s;Start date:%s",schedule.EndDate,schedule.StartDate)) else if not strDate then strDate = string.gsub(schedule.StartDate,"(%w+) .*","%1" ) table.insert( schedulelist,table.remove(schedules,index)) elseif strDate == string.gsub(schedule.StartDate,"(%w+) .*","%1" ) then table.insert( schedulelist,table.remove(schedules,index)) else ---- 没有数据删除index需要++,否则index不变继续遍历 index = index + 1 end end end for index,schedule in ipairs(useingSches) do if strDate == string.gsub(schedule.StartDate,"(%w+) .*","%1" ) then log.info("schedule","Removed using schedule",string.format("End date:%s;Start date:%s",schedule.EndDate,schedule.StartDate)) schedule.EndDate = nil schedule.StartDate = nil table.remove(useingSches, index) ---- NOTE:已经执行的策略删除,会导致策略直接停止,也就是这路处于开灯状态 end end while(#schedulelist > 0) do local schedule = table.remove(schedulelist) log.info("schedule","Added using schedule",string.format("End date:%s;Start date:%s",schedule.EndDate,schedule.StartDate)) table.insert(useingSches,schedule) ----从待生效移动到执行策略 end changed = true end table.sort(useingSches,function(a,b)---- 对运行中的sche排序 return strDate2Time(a.StartDate) < strDate2Time(b.StartDate) end) end if changed then io.writeFile(SCHEDULE_FILE,cjson.encode(scheduleCircuits),"w+") end end end end local function pp2app(topic,payload) local currTime = os.time() -- print("pp2app",topic,payload) local obj = cjson.decode(payload) local tags = obj.tag_node and cjson.decode(obj.tag_node) or nil ---- 非通用方式计算 paser_special(obj,tags) ---- 先对数据做汇聚 realDataCala(obj,tags) ---- 用原始数据做只有原始数据做报警预处理 realAlarmCala(obj,tags) local dataChanged = nil ---- 比对数据是否变化 local function getTolerance(l1_key,l2_key) for i,item in ipairs(dataCollectMap) do local des_l1_key = item[4] local des_l2_key = item[5] local tolerance = item[8] and item[8] or 1 if des_l1_key == l1_key and des_l2_key == l2_key then return tolerance end end end for des_l1_key,item in pairs(realData) do if not HistoryData[des_l1_key] then HistoryData[des_l1_key] = {} end local tmp = HistoryData[des_l1_key] for des_l2_key,value in pairs(item) do ---- 这个默认值赋值会影响第一次是否上报 if not tmp[des_l2_key] then tmp[des_l2_key] = 0 dataChanged = true end -- log.info("tag", tmp[des_l2_key] - value) local tolerance = getTolerance(des_l1_key,des_l2_key) if math.abs(tmp[des_l2_key] - value) >= (tolerance and tolerance or 1) then tmp[des_l2_key] = value dataChanged = true log.info(string.format("Data changed[%s.%s],real,hist,tolerance=", des_l1_key,des_l2_key),tmp[des_l2_key],value,tolerance) end end end -- log.info("=== realData ===", dataChanged,cjson.encode(realData)) local alarmChanged = nil ---- 比对数据是否变化 for key,value in pairs(realAlarm) do if historyAlarm[key] ~= value then historyAlarm[key] = value alarmChanged = true end end -- log.info("=== realAlarm ===", alarmChanged, cjson.encode(realAlarm)) local dev_sn = string.format("%s%s",gw_sn,MYSELF_SN_SUFFIX) if dataChanged or (os.difftime(currTime,last_report) >= REPORT_PERIOD) then last_report = currTime internal_api.service_response(gw_sn,dev_sn,"NULL","SDBTotalState",realData,0) end if alarmChanged then internal_api.event(gw_sn,dev_sn,"NULL","SDBWarning",realAlarm ,0) end end local function sys_monitor() while true do sys.wait(10000) -- local date = os.date("*t",os.time() + TIMEZOEN) ---- 找到今天的策略 log.info("Monitor",string.format("%s %s %s %s %s %s %s %s", os.date("%Y-%m-%d %H:%M:%S",os.time() + TIMEZOEN), "Control", "Feedback","ScheMode","Start","End","Manu","Current")) for _,key in ipairs({"CircuitMain","Circuit1","Circuit2","Circuit3","Circuit4"}) do local data = realData[key] log.info("Monitor",string.format("%s %s %s %s %s %s %s %s", key, data and data.Ctrl and data.Ctrl or "X", data and data.Feedback and data.Feedback or "X", using_job[key] and "Sche" or "Def", start_time[key] and start_time[key] or "XX/XX/XX XX:XX:XX", end_time[key] and end_time[key] or "XX/XX/XX XX:XX:XX", manu_cmd[key] and manu_cmd[key] or "X", data and data.Current and data.Current or "XXX")) end end end local function KM_conflict_check() while true do sys.wait(300) local curr = os.time() for _,key in ipairs({"CircuitMain","Circuit1","Circuit2","Circuit3","Circuit4"}) do local alarm_id = string.format("%s.Feedback",key) if not cacheAlarm[alarm_id] then cacheAlarm[alarm_id] = {false,false,false,false,false,false} end local tmpAlarm = cacheAlarm[alarm_id] local data = realData[key] if data and data.Ctrl and data.Feedback then if data.Ctrl ~= data.Feedback and not conflictFlagTime[key] then conflictFlagTime[key] = os.time() tmpAlarm[ALARM_CONFLICT] = false elseif data.Ctrl == data.Feedback and conflictFlagTime[key] then conflictFlagTime[key] = nil end end local conflictTime = conflictFlagTime[key] -- 这里仅仅更新 cacheAlarm 的标记,不触发报警 if not conflictTime then tmpAlarm[ALARM_CONFLICT] = false else if os.difftime(curr,conflictTime) > 5 then tmpAlarm[ALARM_CONFLICT] = true end end end end end local function handler() local pid, status, code = syswait.wait(-1, syswait.WNOHANG) log.error('system exit', pid, status, code) --打印不出来 os.exit(0) end local function gpioInit() local dio_sn = string.format("%s_DIO",gw_sn) internal_api.call_dev_service(gw_sn,dio_sn,"RS485_3","DO4",{open=0}) internal_api.call_dev_service(gw_sn,dio_sn,"RS485_3","DO5",{open=0}) end function main(nodes_cfg,sn) log.info("main_loop", "=============== Direct Current RUN 1.0.5 ============") gw_sn = sn if not gw_sn then gw_sn = io.getGatewayID() -- gw_sn = "218800000010" end local devs = getSubDevice(gw_sn) log.info("device list:",cjson.encode(devs)) log.info("main_loop", string.format("Try Run At Shell\nsn='%s'\nnodes='%s'\nmain_loop(nodes,sn)",gw_sn or "NOT SN",nodes_cfg or "")) siganl.signal(siganl.SIGKILL,handler) siganl.signal(siganl.SIGHUP,handler) internal_api.local_mqtt_session(devs,"APP",iot2app,pp2app) load_config() gpioInit() schedule_monitor() default_job() sys.taskInit(schedule_merge) sys.taskInit(leak_current) sys.taskInit(control_priority) sys.taskInit(sys_monitor) sys.taskInit(KM_conflict_check) --启动系统框架 sys.init(0, 0) sys.run() end main(arg[1],arg[2])