server/net-service.cpp

#include <thread>
#include <mutex>
#include <atomic>
#include <iostream>
#include <memory>
#include <string.h>
#include <math.h>
#include <log.h>
#include <znet.h>
//#include <zstream.h>
#include <sys/time.h>
#include "worker.h"
#include "tdoa_sync.h"
#include "protocol.h"
#include "net-service.h"
#include "ant.h"
#include "card.h"
#include "crc.h"
#include "mine_business.h"
#include "tool_time.h"
#include "module_service/module_traffic_light_manager.h"

net_service::net_service()
{
	m_loc_worker=worker::instance();
	m_sync_worker=tdoa_sync::instance();
}

net_service::~net_service()
{
	if(m_loc_worker)
		m_loc_worker->stop();
}

void net_service::on_timer()
{
	visit_site_status vss;
	sit_list::instance()->accept(vss);

    traffic_light_manager::instance()->visit_light_status();
}

void net_service::on_connect(const std::shared_ptr<client>& clt)
{
     
}

static bool check_message_time(const std::shared_ptr<client> &clt, const std::shared_ptr<site> &site_ptr, struct tm*p_site_tm, int site_ct, int power)
{
	bool result=true;

	char timebuf[64]{0};
	strftime(timebuf,64,"%F %T",p_site_tm);

	logn_info(1,"分站数据信息:net=%s,site=%d,tm=%s,sct=%d,power=%s", clt->name().c_str(), site_ptr->m_id,timebuf,
			site_ct, site_ptr->m_power_check_enable?((power&1)?"ac":"dc"):"??");

	time_t site_time=mktime(p_site_tm);
	double diff=difftime(site_time, site_ptr->last_site_time());
	if(diff<-6) //允许2秒的时间抖动  3
	{
		logn_error(1,"分站时间回退,数据将被丢弃:net=%s,site=%d,diff=%d",clt->name().c_str(),site_ptr->m_id,(int)diff);
		result=false;
	}

	diff=difftime(site_time, time(nullptr));
	if(fabs(diff)>6) //3
	{
		logn_error(1,"分站时间与服务器时间相差太大:%s,site=%d,diff=%d",clt->name().c_str(),site_ptr->m_id,(int)diff);
		result=false;
	}

	site_ptr->set_site_time(site_time);

	return result;
}

void net_service::on_message(const std::shared_ptr<client> &clt,const char*data,size_t len)
{
	bool message_handled=true;
	try
	{
		zistream is(data,len-2);
		uint16_t cmd;
		is>>skip(2)>>cmd;
		switch(cmd)
		{
			case CHAR_LOCATEDATA_TOF_EXTEND://tof-扩展
				{
					uint32_t site_id=-1;
					uint16_t site_ct=-1;
					uint8_t  power=-1;
					struct tm site_tm={0}; 

					is>>site_id>>site_ct
						>>*(uint8_t*)&site_tm.tm_sec >>*(uint8_t*)&site_tm.tm_min >>*(uint8_t*)&site_tm.tm_hour >>*(uint8_t*)&site_tm.tm_mday
						>>*(uint8_t*)&site_tm.tm_mon >>*(uint8_t*)&site_tm.tm_mon >>*(uint8_t*)&site_tm.tm_year 
						>>skip(2)>>power;

					site_tm.tm_year+=2000-1900;
					site_tm.tm_mon-=1;

					const auto &site_ptr = sit_list::instance()->get(static_cast<int32_t>(site_id));
					if(!site_ptr)
					{
						logn_error(1,"未定义分站：net=%s,site=%d", clt->name().c_str(), site_id);
						break;
					}

                    if(site_ptr->is_abnormal_site())
                    {
						logn_error(1,"分站[%d]天线异常", site_id);
                        break;
                    }

					check_message_time(clt,site_ptr,&site_tm,site_ct,power);
                    site_ptr->m_algo = LDT_TOF;

                    if(clt->type()!=2){
					    site_ptr->set_client(clt);
					    site_ptr->on_power_status((power&1)==0);
                        //clt->set_site_id(site_id);
                    }

					struct timeval tv;
					gettimeofday(&tv,NULL);
					uint64_t t = tv.tv_sec*1000 + tv.tv_usec/1000;
					int index=0;
					while(!is.eof())
					{
						uint64_t tstamp = t -1000 + 50 + index* 45;
						index++;
						task*t=task::alloc<message_locinfo>();
						message_locinfo&m=t->body<message_locinfo>();
                        m.m_loc_type = LDT_TOF;
						m.m_site_id=site_id;
						m.m_time_stamp=tstamp;
						m.load(is,false);
						//t_site->m_site_data = 0;

						t->m_cmd_code=cmd;
						t->m_hash_id=m.m_card_id;
						m_loc_worker->request(t);
					}
				}
				break;
			case CHAR_LOCATEDATA_TOF_OPTIMIZE://tof-优化,大小分站
				{
					uint16_t site_id=-1;
					uint16_t site_ct=-1;
					uint32_t site_time_s=0;
					uint8_t  power=-1;

					is>>site_id>>site_ct >>site_time_s >>power;
					power=power>>6;

					time_t site_time=site_time_s;

					auto site_ptr = sit_list::instance()->get(static_cast<int32_t>(site_id));
					if(!site_ptr)
					{
						logn_error(1,"在全局分站列表中找不到分站：%s,%d", clt->name().c_str(), site_id);
						break;
					}

                    if(site_ptr->is_abnormal_site())
                    {
						logn_error(1,"分站[%d]天线异常", site_id);
                        break;
                    }

					struct tm site_tm={0};
					localtime_r(&site_time, &site_tm);
					check_message_time(clt,site_ptr,&site_tm,site_ct,power);

                    if(clt->type()!=2){
					    site_ptr->set_client(clt);
					    site_ptr->on_power_status((power&1)==0);
                        //clt->set_site_id(site_id);
                    }

					struct timeval tv;
					gettimeofday(&tv,NULL);
					uint64_t t = tv.tv_sec*1000 + tv.tv_usec/1000;
					int index=0;
					while(!is.eof())
					{
						uint64_t tstamp = t -1000 + 50 + index* 45;
						index++;
                        
						for(auto & tk : message_locinfo::load_opt(is))
						{
							message_locinfo&m=tk->body<message_locinfo>();
							m.m_site_id=site_id;
							m.m_time_stamp=tstamp;

							tk->m_hash_id=m.m_card_id;
							tk->m_cmd_code=cmd;

							m_loc_worker->request(tk);
						}
					}
				}
				break;
            case CHAR_LOCATEDATA_PDOA:
                {
                    int32_t site_id = parse_data_anchor_opt(clt, is);
                    if(site_id < 0){
                        break;
                    }
                    const auto& site_ptr = sit_list::instance()->get(static_cast<int32_t>(site_id));
                    if(!site_ptr){
                        logn_error(1,"在全局分站列表中找不到分站：%d", site_id);
                        break;
                    }                    
                    
					if(clt->type() != 2){
					    site_ptr->set_client(clt);
					    //site_ptr->on_power_status((power&1)==0);
                    }

                    site_ptr->set_algo(LDT_PDOA);
                    //log_info("[scale_test] %.2f", site_ptr->m_scale);
                    struct timeval tv;
                    gettimeofday(&tv, NULL);
                    uint64_t cur_time = tv.tv_sec*1000 + tv.tv_usec/1000;
                    int index = 0;
                    while(!is.eof()){
                        uint64_t tstamp = cur_time - 1000 + 50 + index*45;
                        ++index;
                        task* t = task::alloc<message_pdoa_locinfo>();
                        message_pdoa_locinfo& m = t->body<message_pdoa_locinfo>();
                        m.load(is);
                        float pdoa = 10.0;
                        float dist = m.m_tof*15.65*2.996*1e-4;
                        m.m_site_id = site_id;  
                        m.m_time_stamp = tstamp;
                        m.m_loc_type = LDT_PDOA;
                        t->m_cmd_code = cmd;
                        t->m_hash_id = m.m_card_id;
                        pdoa = tool_other::get_pdoa(m.m_poa, site_ptr->m_pdoa_offset);
                        logn_info(3, "[pdoa] site_id=%d, card_id=%d, ct=%d, dist=%.3f, rav=%d, poa1=%.4f, poa2=%.4f, poa3=%.4f, pdoa=%.4f, pdoa_offset=%.4f", site_id, m.m_card_id, m.m_card_ct, dist, m.m_rav, m.m_poa[0], m.m_poa[1], m.m_poa[2], pdoa, site_ptr->m_pdoa_offset);
                     
                        if(m.m_poa[1] == 10.0 || dist < 0.0001 ){
                            continue;
                        }
                        m_loc_worker->request(t);
                    }
                }
                break;
			case CHAR_NET_CALI_TIME://tof 分站时间同步
				{
					// 网络设备校时
					net_cali_time(clt);
				}
				break;
			case CHAR_CAN_CALI_TIME:
				{
					// can设备校时
					uint32_t rid=0;
					is>>rid;
					can_cali_time(clt, rid);
				}
				break;
			case CHAR_LOCATEDATA_TDOA_EXTEND:       // 0x863b
            case CHAR_LOCATEDATA_TDOA_EXTEND_INS:   // 0x901b
				{
                    int32_t site_id = parse_data_anchor(clt, is);
					if(site_id < 0){
                        break;
                    }
                    const auto& site_ptr = sit_list::instance()->get(static_cast<int32_t>(site_id));
                    if(!site_ptr){
                        logn_error(1,"在全局分站列表中找不到分站：%d", site_id);
                        break;
                    }
                    site_ptr->set_algo(LDT_TDOA);

					while(!is.eof())
					{
						task* t = task::alloc<message_tdoa_locinfo>();
						message_tdoa_locinfo& m = t->body<message_tdoa_locinfo>();
                        m.m_loc_type = LDT_TDOA;
						m.load(is, cmd);
						t->m_cmd_code = cmd;
						t->m_hash_id = m.m_card_msg.m_id;
						m_sync_worker->translate(m);
						m_loc_worker->request(t);
					}
				}
				break;
			case CHAR_TDOA_READER_SYNC_TIME://time sync 0xa78d
				{
					message_tdoasync m;
					m.load(is);
					m_sync_worker->on_message(m);
				}
				//site_message::on_sync(this,t.m_param1);
				break;
            case CHAR_VIRTUAL_DATA_PUSH_CMD://虚拟数据链接
                {
                    logn_info(1,"接收到虚拟链接:%s",clt->name().c_str());
                    clt->set_conn_type(2);
                }
                break;

            case CHAR_READER_HEART:
                { 
					//通信基站心跳    
					uint32_t id = 0;
                    uint16_t site_ct = 0;
                    uint16_t reserve = 0;
                    is>>id>>site_ct>>reserve;

                    auto site_ptr = sit_list::instance()->get(id);
                    if(!site_ptr){
                        logn_error(1,"在全局分站列表中找不到分站：%d", id);
                        break;
                    }

					site_ptr->m_device_type_id = 0x08;
                }
                break;
            case 0x793B: //虚拟推送的心跳测试
			case CHAR_LOCATEDATAHIS_TOF_EXTEND://tof his
			case CHAR_LOCATEDATAHIS_TDOA_EXTEND://tdoa his
			case CHAR_CTRL_READER_CMD://ctrl site message
                break;
            case THIRD_PARTY_CHAR_LIGHT_REQ_STATE:
                {
                    // 红绿灯请求状态指令
                    uint32_t id = 0;
                    uint8_t dev_type = 0;

                    is>>id>>dev_type;
                    logn_info(4, "[traffic_light] light request state, light_id=%d, dev_type=%d", id, dev_type);
                    // 找灯，并下发指令
                    auto light_ptr = traffic_light_manager::instance()->get(id);
                    if(light_ptr)
                    {
                        //traffic_light_manager::instance()->send_light_ctrl(id, DT_LIGHT, light_ptr->m_state); 
                        traffic_light_manager::instance()->send_light_data(id, DT_LIGHT, light_shape::green_all_on);
                    }else{
                        traffic_light_manager::instance()->send_light_data(id, DT_LIGHT, light_shape::green_spark);
                    }

                }
                break;
            case THIRD_PARTY_CHAR_LIGHT_HEART:
                {
                    // 红绿灯心跳
                    uint32_t id = 0;
                    uint16_t stamp = 0;
                    uint8_t status = 0;
                    is>>id>>stamp>>status;
                    logn_info(4, "[traffic_light] light heart message, light_id=%d, stamp=%d, status=%d", id, stamp, status);
                    logn_info(1, "红绿灯数据 : light_id=%d, ct=%d", id, stamp);

                    auto light_ptr = traffic_light_manager::instance()->get(id);
                    if(!light_ptr)
                    {
                        logn_error(1,"在全局信号灯列表中找不到信号灯：%d", id);
                        break;
                    }
                    
                    //log_info("[traffic_light] light client type=%d", clt->type());
                    if(clt->type() != 2){
                        light_ptr->set_client(clt);
                    }

                    light_ptr->m_state = status;
                    light_ptr->m_rec_time = time(0);

                    /*if((stamp%10) == 0){
                        int shape = (status == light_shape::red_spark)?light_shape::green_spark:light_shape::red_spark;
                        traffic_light_manager::instance()->send_light_ctrl(id, 0x05, shape);
                    }*/
                }
                break;
			default:
				message_handled=false;
		}

	}
	catch(const std::exception&e)
	{
		logn_error(1,"分站数据处理失败，将关闭分站连接:%s [%s]",clt->name().c_str(),e.what());
		clt->close();
	}

	if(!message_handled)
	{
		logn_error(1,"分站数据未被处理，site=%s",clt->name().c_str());
	}
}

int32_t net_service::parse_data_anchor(const std::shared_ptr<client>& clt, zistream& s)
{
    uint32_t site_id=-1;
    uint16_t site_ct=-1;
    uint8_t  power=-1;
    struct tm site_tm={0}; 

    // 4字节分站号
    // 2字节时间戳
    // 7字节分站时间
    // 2字节保留字节
    // 1字节电源状态
    s>>site_id>>site_ct
        >>*(uint8_t*)&site_tm.tm_sec >>*(uint8_t*)&site_tm.tm_min >>*(uint8_t*)&site_tm.tm_hour >>*(uint8_t*)&site_tm.tm_mday
        >>*(uint8_t*)&site_tm.tm_mon >>*(uint8_t*)&site_tm.tm_mon >>*(uint8_t*)&site_tm.tm_year 
        >>skip(2)>>power;

    site_tm.tm_year+=2000-1900;
    site_tm.tm_mon-=1;

    const auto &site_ptr = sit_list::instance()->get(static_cast<int32_t>(site_id));
    if(!site_ptr)
    {
        logn_error(1,"未定义分站：net=%s,site=%d", clt->name().c_str(), site_id);
        return -1;
    }

    if(site_ptr->is_abnormal_site())
    {
        logn_error(1,"分站[%d]天线异常", site_id);
        return -1;
    }

    check_message_time(clt, site_ptr, &site_tm, site_ct, power);

    if(clt->type()!=2){
        site_ptr->set_client(clt);
        site_ptr->on_power_status((power&1)==0);
    }

    logn_info(1,"分站数据：site_id: %d, site_ct: %d, time: %s, power: %d", site_id, site_ct, tool_time::to_str_ex(&site_tm).c_str(), power);

    return site_id;
}

// 分站优化协议
int32_t net_service::parse_data_anchor_opt(const std::shared_ptr<client>& clt, zistream& s)
{
    // 分站号，2字节
    uint16_t site_id = 0;
    // 分站时间戳，2字节
    uint16_t site_ct = 0;
    // 分站时间，4字节
    uint32_t site_time_stamp = 0;
    // 大小分站所属关系，1字节
    uint8_t reverse = 0;
    // 状态字节，1字节，包括3部分：供电状态，设备类型，保留
    uint8_t power = 0;
    struct tm site_tm={0}; 

    s>>site_id>>site_ct>>site_time_stamp>>reverse>>power;

    time_t _sts = site_time_stamp;
    localtime_r(&_sts, &site_tm);

    const auto &site_ptr = sit_list::instance()->get(static_cast<int32_t>(site_id));

    if(!site_ptr)
    {
        logn_error(1,"未定义分站：net=%s, site=%d", clt->name().c_str(), site_id);
        return -1;
    }

    // pdoa分站不检查天线是否相等
    /*if(site_ptr->is_abnormal_site())
    {
        logn_error(1,"分站[%d]天线异常", site_id);
        return -1;
    }*/

   	site_ptr->m_device_type_id = (power>>3)&0x07;

    if(!check_message_time(clt, site_ptr, &site_tm, site_ct, power)){
		// 分站时间异常，直接校时
		switch(site_ptr->m_device_type_id)
		{
		case 0:
			can_cali_time(clt, site_id);
			break;
		case 1:
			net_cali_time(clt);
		default:
			net_cali_time(clt);
			break;
		}
	
		return -1;
	}

    if(clt->type()!=2){
        site_ptr->set_client(clt);
        site_ptr->on_power_status((power&1)==0);
    }

	uint64_t tt = site_time_stamp;
    tt *= 1000;
    logn_info(1,"分站数据：site_id: %d, site_ct: %d, time_t: %llu, time: %s, power: %d, device_type: %d", site_id, site_ct, tt, tool_time::to_str_ex(tt).c_str(), power, site_ptr->m_device_type_id);

    return site_id;
}

void net_service::net_cali_time(const std::shared_ptr<client>& clt)
{
	// 从第一个字节开始，分别表示毫秒（2字节）、秒、分、时、天、月、年
	unsigned char buf[20]={0,13,0x78,0x3b};

	struct timeval tv;
	gettimeofday(&tv,0);

	struct tm buff={0};
	const struct tm*t=localtime_r(&tv.tv_sec,&buff);

	int p=4;
	buf[p++]=(tv.tv_usec/1000)&0xFF;
	buf[p++]=((tv.tv_usec/1000)>>8)&0xFF;

	buf[p++]=t->tm_sec+1;	// 由于硬件扫描方式会慢1秒，给补偿1秒
	buf[p++]=t->tm_min;
	buf[p++]=t->tm_hour;
	buf[p++]=t->tm_mday;
	buf[p++]=t->tm_wday;
	buf[p++]=t->tm_mon+1;
	buf[p++]=t->tm_year%100;

	uint16_t ccrc=do_crc(buf+2,11);

	buf[p++]=ccrc>>8;
	buf[p++]=ccrc&0xff;

	std::vector<char> tmp((char*)buf,(char*)buf+15);
	clt->send(std::move(tmp));

	logn_info(1,"分站时间同步:ip=%s,time=%d-%02d-%02d %02d:%02d:%02d.%03d",
			clt->name().c_str(),buf[12]+2000,buf[11],buf[9],buf[8],buf[7],buf[6],buf[5]*256+buf[4]);

}

void net_service::can_cali_time(const std::shared_ptr<client>& clt, const int& reader_id)
{
	unsigned char buf[30]={0,17,0x78,0x4b};

	struct timeval tv;
	gettimeofday(&tv,0);

	struct tm buff = {0};
	const struct tm *t = localtime_r(&tv.tv_sec,&buff);

	int p=4;
	for(int i = sizeof(uint32_t) - 1; i >= 0; --i)
	{
		buf[p++] = ((reader_id>>(i*8))&0xFF);
	}
	buf[p++]=(tv.tv_usec/1000)&0xFF;
	buf[p++]=((tv.tv_usec/1000)>>8)&0xFF;

	buf[p++]=t->tm_sec;
	buf[p++]=t->tm_min;
	buf[p++]=t->tm_hour;
	buf[p++]=t->tm_mday;
	buf[p++]=t->tm_wday;
	buf[p++]=t->tm_mon+1;
	buf[p++]=t->tm_year%100;

	uint16_t ccrc=do_crc(buf+2,15);

	buf[p++]=ccrc>>8;
	buf[p++]=ccrc&0xff;

	std::vector<char> tmp((char*)buf,(char*)buf+19);
	clt->send(std::move(tmp));

	logn_info(1,"分站时间同步:site_id=%d, ip=%s, p=%d, time=%d-%02d-%02d %02d:%02d:%02d.%03d",
			reader_id, clt->name().c_str(), p, buf[16]+2000, buf[15], buf[13], buf[12], buf[11], buf[10], buf[9]*256+buf[8]);
}