server/mine_business.cpp

#include <cmath>
#include <sstream>
#include <vector>
#include <list>
#include <array>
#include <map>
#include "mine_business.h"
#include "log.h"
#include "point.h"
#include "db/db_api/CDBConnPool.h"
#include "db/db_tool.h"
#include "common_tool.h"
#include "ant.h"

struct card_sensor
{
	card_sensor(int32_t wid,const point &p,const std::string &cid)
		:card_id(cid)
		,sid(0)
		 ,id(wid)
		 ,bigger(-1),count(0),sum(0),ct(-1),stime(0),flag(false)
	{
		base_point=p;
	}
	std::string			card_id;
	int32_t				sid;
	int32_t				id;
	std::vector<int>	sensorId_list;
	point				base_point;	
	std::array<point,2> ant;
	int8_t bigger;
	uint32_t count,sum;
	uint16_t ct;
	uint64_t stime;
	bool flag;
	std::list<double> his_data;
	std::array<uint64_t,2> dist;
	void clear()
	{
		count=sum=0;stime=0;flag=false;
	}
	bool timeout()
	{
		time_t t=time(NULL);
		return flag && t-stime>10*60;
	}
	bool check_location_reversal(std::string & str)
	{
		time_t t=time(NULL);
		time_t tval= t-stime;
		bool f=false;
		if (tval !=0 && sum*1.0/tval > 0.8 && count*1.0/sum <0.2)
		  f=true;
		char buf[1024]={0};
		sprintf(buf,"[reverse_alarm_checking]%s,count:%d,sum:%d,val:%ld,s:%.2f,v:%.2f,alarm:%d",
					card_id.c_str(),count,sum,tval,(sum*1.0/tval),(count*1.0/sum),f);
		str=buf;
		return f;
	}
	void make_reverse_condition(int16_t antid,uint16_t ct,double d)
	{
		dist[antid]=d;
		if(ct == ct){
			sum++;
			int b=-1;
			if (dist[0]>dist[1]) b=0 ;else b =1;
			if(b==bigger) count++; //合格的点
		}
		ct = ct;
	}
	void set(const point &p0,const point &p1,int id)
	{
		int8_t b=0;
		ant[0].set(p0);ant[1].set(p1);
		double d0=base_point.dist(p0);double d1=base_point.dist(p1);
		if(d0>d1)b=1;else b=0;
		if (bigger!=-1 && b!=bigger){
			count=sum=0;stime=time(0);
		}
		bigger=b;sid=id;
	}
	bool push(double d){
		bool f=true;
		his_data.push_back(d);
		if (his_data.size()==3){
			double da=0;
			for (const auto &d:his_data){
				if(da>d || d-da<0.004){f=false;break;}
				da=d;
			}
			his_data.pop_front();
		}
		else f=false;
		return f;
	}
	point &operator[](int i){return ant[i];}
	const point &operator[](int i)const {return ant[i];}
	void set_true(){flag=true;stime=time(NULL);}
	bool warning(){return flag;}
	bool empty(){return his_data.empty();}
	double back(){return his_data.back();}
	void clear_sensor(){clear();}
	int reader_id(){return sid;}
	void push_back(int32_t sensor_id){sensorId_list.push_back(sensor_id);}
};
struct card_sensor_mgr
{
	void push(int32_t wid,int32_t sid,const point &p,const std::string & cardid)
	{
		std::shared_ptr<card_sensor> cs=nullptr;
		auto it=m_mgr.find(wid);
		if(it!=m_mgr.end())
		  cs=it->second;
		else
		{
			cs=std::make_shared<card_sensor>(wid,p,cardid);
			m_mgr.insert({wid,cs});
		}
		cs->push_back(sid);
	}
	void for_each();
	void make_condition(int32_t antid,uint32_t ct,uint64_t tof,uint32_t sid);
	std::map<int32_t,std::shared_ptr<card_sensor>> m_mgr;
};

struct staffer_num_business
{
	void record_staffer_num();
	void fetch_add(){m_staff_num++;}
	private:
	void reset(){m_staff_num=0;}
	std::atomic<uint32_t> m_staff_num;
	std::time_t m_record_staffer_timeval;
};

struct reverse_alarm_business
{
	void load_sensor();	
	void run_sensor();
	void make_condition(uint64_t type,uint32_t id,int32_t antid,uint32_t ct,uint64_t tof,uint32_t sid);
	private:
	std::map<uint64_t,std::shared_ptr<card_sensor_mgr>> m_sensor_mgr;
};

mine_business::mine_business()
{
	m_staffer_num_ptr.reset(new staffer_num_business);
	m_reverse_alarm_ptr.reset(new reverse_alarm_business);
}
mine_business* mine_business::inst()
{
	static mine_business ct;
	return &ct;
}
void mine_business::run_business()
{
	mine_business::inst()->record_staffer_num();
	mine_business::inst()->handle_reverse_alarm();
}
void mine_business::fetch_add()
{
	m_staffer_num_ptr->fetch_add();
}
void mine_business::record_staffer_num()
{
	m_staffer_num_ptr->record_staffer_num();
}
void mine_business::load()
{
	m_reverse_alarm_ptr->load_sensor();	
}
void mine_business::handle_reverse_alarm()
{
	m_reverse_alarm_ptr->run_sensor();
}
void mine_business::make_reverse_condition(uint64_t type,uint32_t id,int32_t antid,uint32_t ct,uint64_t tof,uint32_t sid)
{
	m_reverse_alarm_ptr->make_condition(type,id,antid,ct,tof,sid);
}
///////staffer_num_business
/**********************************
//每两分钟把人员数量入库。
//不往前端推送的人员，不算在数量范围内
*********************************/
void staffer_num_business::record_staffer_num()
{
	static uint32_t min_num=0,max_num=0;
	std::time_t t = time(NULL);
	char ti[64] = { 0 };
	strftime(ti,sizeof(ti),"%Y/%m/%d %H:%M:%S",localtime(&t));
	std::string sti(ti);
	int minute = atoi(sti.substr(sti.find_first_of(':')+1,2).c_str());
	if(m_record_staffer_timeval==0)
	{
		min_num=max_num=m_staff_num;
		m_record_staffer_timeval = t;
	}
	if(t-m_record_staffer_timeval>=120 && minute%2 == 0)
	{
		std::stringstream ss;
		ss<< "INSERT INTO his_staff_number(max_num,min_num,ave_num) VALUES("<<max_num<<','<<min_num<<','<<std::lround((max_num+min_num)/2)<<");";
		logn_info(2,"staff_number:%s",ss.str().c_str());
		db_tool::PushAsync(ss.str().c_str());

		min_num=max_num=m_staff_num;
		m_record_staffer_timeval = t;
	}
	else
	{
		if (min_num>m_staff_num)   min_num = m_staff_num;
		if (max_num < m_staff_num) max_num = m_staff_num;
	}
	reset();
	return;
}
//前端修改数据库表，dat_sensor dat_drivingface,dat_drivingface_vehicle的时候，需要加载.
void reverse_alarm_business::load_sensor()
{
	std::string sql = "SELECT sensor_id,data_source,work_face_id,v.vehicle_id,ve.card_id,d.base_point_x,d.base_point_y FROM dat_sensor ,dat_drivingface_vehicle v,dat_drivingface d,dat_vehicle_extend ve WHERE sensor_type_id = 1 AND work_face_id = v.drivingface_id AND work_face_id = d.drivingface_id AND ve.vehicle_id = v.vehicle_id;";

	std::string Error;
	YADB::CDBResultSet DBRes;
	sDBConnPool.Query(sql.c_str(),DBRes,Error);
	int nCount = DBRes.GetRecordCount( Error );
	if (nCount < 1)
	  log_error("init_sensor..failed[%s]", sql.c_str());

	while ( DBRes.GetNextRecod(Error) )
	{
		int32_t sensor_id  = 0;
		DBRes.GetField( "sensor_id",sensor_id, Error );

		int32_t data  = 0;
		DBRes.GetField( "data_source",data, Error );

		int32_t workfaceid  = 0;
		DBRes.GetField( "work_face_id",workfaceid, Error );

		std::string card_id;
		DBRes.GetField( "card_id",card_id, Error );
		double bx,by;
		DBRes.GetField( "base_point_x",bx, Error );
		DBRes.GetField( "base_point_y",by, Error );

		log_info("init sensor :workfaceid:%d,sensor_id:%d,data:%d,card:%s,(%.2f,%.2f)",workfaceid,sensor_id,data,card_id.c_str(),bx,by);
		if(data <= 0) continue;
		uint64_t cid = tool_other::card_id_to_u64(card_id);

		std::shared_ptr<card_sensor_mgr> tm=nullptr;
		auto iter = m_sensor_mgr.find(cid);
		if(iter != m_sensor_mgr.end())
		  tm=iter->second;
		else
		{
			tm = std::make_shared<card_sensor_mgr>();
			m_sensor_mgr.insert({cid,tm});
		}
		point p(bx,by);
		tm->push(workfaceid,sensor_id,p,card_id);
	}
}

void reverse_alarm_business::run_sensor()
{
	static time_t time_interval=0;
	time_t t = time(NULL);
	if (t-time_interval>3*60)
	{
		time_interval =t;
		printf("reverse_alarm......\n");
		for(const auto &x:m_sensor_mgr)
			x.second->for_each();
	}
}

void card_sensor_mgr::for_each()
{
	for(const auto & cst:m_mgr)
	{
		auto cs = cst.second;
		if(cs->sensorId_list.empty())
		  continue;
		char strsql[512]={0};
		snprintf(strsql,512, "SELECT data_value FROM his_sensor_data WHERE sensor_id = %d ORDER BY write_time DESC LIMIT 3;",cs->sensorId_list[0]);
		std::string Error;
		YADB::CDBResultSet DBRes;
		sDBConnPool.Query(strsql,DBRes,Error);
		int nCount = DBRes.GetRecordCount( Error );
		if (nCount < 1)
		{
		  log_error("load_sensor_data..failed[%s]", strsql);
		  continue;
		}

		double prenum=-1;
		double nrow=0;double sum=0;bool f=false;
		while ( DBRes.GetNextRecod(Error) )
		{
			double num  = 0;
			DBRes.GetField( "data_value",num, Error );

			if (point::eq(num,0,1e-10))continue; //0异常数据抛掉
			if(point::eq(prenum,-1,1e-10)){ //如果没有参考数据，则判断，如果有历史数据，用历史，没有则用第一个数据。
				if(cs->empty()) prenum=num; else prenum=cs->back();
			}
			if (prenum-num>0.1){ //异常数据 如果之前存在正常数据则继续，如果没有则本次不作参考。0.1暂定
				if(cs->empty()){
					f=true;break;
				}
				else continue;
			}
			nrow++;sum+=num;		
		}
		if (!f && nrow!=0){
			double result=sum/nrow;
			log_info("reverse_alarm_push %s,%.3f",cs->card_id.c_str(),result);
			if(cs->push(result)){//满足三个并且呈现上升趋势
				log_info("reverse_alarm_up %s",cs->card_id.c_str());
				if (!cs->warning())
				  cs->set_true();//置true,主进程开始抓取数据。
			}
		}
		if (cs->timeout()){//时间满足，判断是否告警
			auto reader_it = sit_list::instance()->get(cs->reader_id());
			if (!reader_it){
				cs->clear_sensor();
				continue;
			}
			std::string msg;
			if(cs->check_location_reversal(msg))//满足告警条件，查看是否已经存在告警，没有产生告警
			{
				log_info("reverse_alarm_create:%s",cs->card_id.c_str());
				//产生告警。
				//deal_readers_event(ET_READER_LOCATION_REVERSAL,true,reader_it->second);
			}
			else  //不满足告警条件，查看是否已经存在告警，存在则取消告警。
			{
				log_info("reverse_alarm_delete:%s",cs->card_id.c_str());
				//补
				//deal_readers_event(ET_READER_LOCATION_REVERSAL,false,reader_it->second);
			}
			log_info("%s",msg.c_str());
			cs->clear_sensor();//清除数据，重新判断
		}
	}
}
void reverse_alarm_business::make_condition(uint64_t type,uint32_t id,int32_t antid,uint32_t ct,uint64_t tof,uint32_t sid)
{
	uint64_t cid = tool_other::type_id_to_u64(type,id);
	auto it = m_sensor_mgr.find(cid);
	if(it != m_sensor_mgr.end())
		it->second->make_condition(antid,ct,tof,sid);
}
void card_sensor_mgr::make_condition(int32_t antid,uint32_t ct,uint64_t tof,uint32_t sid)
{
	for(const auto x:m_mgr)
	{
		auto pdc =x.second;
		if (pdc->warning()){
			auto sit = sit_list::instance()->get(sid);
			point p0=(*sit)[0];point p1=(*sit)[1];
			if (!((*pdc)[0]==p0 && (*pdc)[1]==p1)){
				pdc->set(p0,p1,sid); 
				log_info("[reverse_alarm_start]%s,bigger:%d",pdc->card_id.c_str(),pdc->bigger);
			}
			log_info("[reverse_alarm_working]%s,ct:%d,antid:%d,dist:%.2f",pdc->card_id.c_str(),ct,antid,tof);
			pdc->make_reverse_condition(antid,ct,tof);
		}
	}
}