export LD_LIBRARY_PATH=<YOUR DolphinDB Dir>/server/plugins/rtseis/:$LD_LIBRARY_PATH
export LD_LIBRARY_PATH=<YOUR DolphinDB Dir>/server/plugins/tf/:$LD_LIBRARY_PATH

try{ loadPlugin("./plugins/filterpicker/PluginFilterPicker.txt") }catch(ex){print(ex) }
try{ loadPlugin("./plugins/rtseis/PluginRTSeis.txt") }catch(ex){print(ex) }
try{ loadPlugin("./plugins/tf/PluginTf.txt") }catch(ex){print(ex) }

unsubscribeTable(tableName = "dataStream", actionName="filterPickerPredict");go
unsubscribeTable(tableName = "pickerStream",actionName="tensorFlowPredict");go
try{dropStreamTable("dataStream")}catch(ex){print(ex)}
try{dropStreamTable("pickerStream")}catch(ex){print(ex)}
try{dropStreamTable("tensorStream")}catch(ex){print(ex)}

  //dataStream，接收波形实时流数据
enableTableShareAndPersistence(table=streamTable(1000000:0, `tagid`ts`data, [INT,TIMESTAMP, INT]), tableName=`dataStream, cacheSize=1000000);
  //pickerStream
enableTableShareAndPersistence(table=streamTable(100000:0, `ts`id, [TIMESTAMP,INT]), tableName=`pickerStream, cacheSize=1000000);
  //tensorStream
enableTableShareAndPersistence(table=streamTable(100000:0, `ts`id, [TIMESTAMP, INT]), tableName=`tensorStream, cacheSize=1000000);

if(existsDatabase("dfs://seis01")) dropDatabase("dfs://seis01");
  //创建分布式数据库
create database "dfs://seis01" partitioned by VALUE(1..10), engine='TSDB'
  //创建维度表
create table "dfs://seis01"."tagInfo"(
	xfdsn SYMBOL,
	net SYMBOL,
	sta SYMBOL,
	loc SYMBOL,
	chn SYMBOL,
	id INT[compress="delta"]
)
sortColumns=[`id];
  //向维度表插入数据
net = ["ZJ","YN","XZ","XJ","TJ"]
sta = ["A0001","A0002","A0003","A0004","A0005","A0006","B0001","B0002","B0003","C0001"]
tmp = `EIE`EIN`EIZ
netList = stretch(net,150)
staList = take(stretch(sta,30),150)
locList = take(`40,150)
chn = take(tmp,150)
colt =   array(STRING)
for(i in 0..(chn.size()-1)){
	colt.append!( chn[i].split()[0] + "_" + chn[i].split()[1] + "_" +chn[i].split()[2] )
}
xfdsn = "XFDSN:"+netList+"_"+staList+"_"+locList+"_"+colt
t = table(xfdsn,netList as net,staList as sta,locList as loc,chn,1..150 as id)
loadTable( "dfs://seis01","tagInfo").append!(t);

/*
 * 模拟实时数据
 */
def insertIntoDataStream(){
	do{
		tagidList = 1 2 3 4 5 6 7 8 9
		ts = now()+(0..400)*10
		t = table(stretch(tagidList,3600)as tagid,take(ts,3600) as ts,randNormal(-2500, 1000, 3600) as data)
		objByName(`dataStream).append!(t)
		sleep(3500)
	}while(true)
}
jobId = submitJob("simulate","simulate",insertIntoDataStream);
//通过以下方式取消Job
//cancelJob(jobId)

/*
 *一级异常检测处理函数
 *计算波形实时数据中的突峭时间点
 *返回值为突峭时间点和通道id
 */
def pickerHandler(mutable pickerSt, mutable res, vz, msg){
	tags=groups(msg[`tagid]) //按tagid分组
	for (tag in tags.keys()){
		re=res[tag]
		if(re==NULL){
			//创建filter handler
			re= filterPicker::createFilterPicker()
			res[tag]=re
		}
		//将数据传入filterPicker进行计算
		vt=filterPicker::filterPicker(re,msg[tags[tag]][`ts], msg[tags[tag]][`data].float(), 1000)
		if(vt.size()>0){
			vid=take(tag,vt.size())
			pickerSt.append!(table(vt as ts,vid as id))
		}
	}
}

vz=exec id from loadTable("dfs://seis01","tagInfo") where chn ilike("%z") //只需要计算Z分量的数据
res=dict(INT, ANY)
//订阅dataStream，进行一级异常检测，异常检测结果输出到filterStream中
subscribeTable(tableName = "dataStream", actionName="filterPickerPredict", offset=-1,handler=pickerHandler{objByName("pickerStream"), res, vz, }, msgAsTable=true, batchSize = 2000, throttle=1,reconnect=true)

def tensorHandler(mutable tensorSt, mutable infoTable, mutable model, mutable cache, msg){
	if(msg.type()!=0) cache.append!(msg)	
	if(cache.size()==0) return
	currTime = now() - 2000
	outed = select * from cache where timestamp  < currTime
	for(s in outed){
		//获取同个台站id信息
		info = select net,sta,loc from infoTable where id == s[`id]
		netInfo = info[`net][0]
		staInfo = info[`sta][0]
		bhzInfo = info[`loc][0]
		bheId = exec id from infoTable where net == netInfo and sta == staInfo and loc == bhzInfo and ilike(chn,"%e")
		bhnId = exec id from infoTable where net == netInfo and sta == staInfo and loc == bhzInfo and ilike(chn,"%n")
		bhzId = exec id from infoTable where net == netInfo and sta == staInfo and loc == bhzInfo and ilike(chn,"%z")
		//获取三个分量数据
		begT = timestamp(s[`timestamp] - 2000)
		endT = timestamp(s[`timestamp] + 1999)
		ve = exec data from dataStream where tagid ==bheId[0] and ts between(begT:endT)
		vn = exec data from dataStream where tagid ==bhnId[0] and ts between(begT:endT)
		vz = exec data from dataStream where tagid ==bhzId[0] and ts between(begT:endT)
		if(ve.size()==  400 && vn.size() ==  400 && vz.size() ==  400) {
			//去趋势
			tem = ols(vn, 0..399)
			bhn = vn - (tem[0] + tem[1]*0..399)
			tem = ols(ve, 0..399)
			bhe = ve - (tem[0] + tem[1]*0..399)
			tem = ols(vz, 0..399)
			bhz = vz - (tem[0] + tem[1]*0..399)
			//滤波
			bhn = rtseis::sosBandPass(bhn,4,3.0,20.0,0,100.0,0.0);
			bhe = rtseis::sosBandPass(bhe,4,3.0,20.0,0,100.0,0.0);
			bhz = rtseis::sosBandPass(bhz,4,3.0,20.0,0,100.0,0.0);
			m = matrix(bhn,bhe,bhz)
			//矩阵数据归一化
			m1 = m / max(max(abs(m)))
			//输入模型进行预测, ts返回值为预测结果
			ts = tf::predict(model, float(m1))
			//prob_p 是P波的概率，prob_S是S波概率, prob_N是噪声的概率。
			prob_P = ts[0]
			prob_S = ts[1]
			prob_N = ts[2]
			if(prob_P > 0.90 || prob_S > 0.90){
				tensorSt.append!(table(s[`timestamp] as timestamp, s[`id] as id))
			}	
		 }
	}
	delete from cache where timestamp  < currTime	
}

  //导入模型
model = tf::load("./model_pol.pb", "conv1d_1_input", "activation_7_1/concat")
infoTable = select * from  loadTable("dfs://seis01","tagInfo")
vz=exec id from loadTable("dfs://seis01","tagInfo") where chn ilike("%z") //只需要计算Z分量的数据
cache =   table(10:0, `timestamp`id, [TIMESTAMP, INT])
  //订阅pickerStream，进行二级异常检测
subscribeTable(tableName = "pickerStream",actionName="tensorFlowPredict", offset=-1, handler=tensorHandler{objByName("tensorStream"),infoTable, model,cache,}, msgAsTable=true, batchSize = 1000, throttle=0.4, timeTrigger=true,reconnect=true)