修改

Home/Lib/Action/RouteRfidKafkaAction.class.php

@@ -18,35 +18,55 @@ class RouteRfidKafkaAction extends Action {
 			return array('success'=>false,'message'=>'addRfidDataToNingbo failed,labels not existed!');
 		}
 		
-		$conn = null;
+		
 		$host= '61.175.203.188';
 		$port=   '6521';
 		$instance_name= 'DSSC';
 		$username=  'dssc3';
 		$password=  'dssc3';
-		$conn = new PDO("oci:dbname=//".$host.":".$port."/".$instance_name,$username,$password);// PDO方式
-		$conn->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
+		$conn = oci_connect($username, $password, $host.':'.$port.'/'. $instance_name,'AL32UTF8');
+		if (!$conn) {
+		   $e = oci_error();
+		   trigger_error(htmlentities($e['message'], ENT_QUOTES), E_USER_ERROR);
+		}
+		$RF_ID=strtoupper($data['mac']);
+		$station_sql='SELECT DEVICE_NAME FROM DSSC2.ADM_DEV WHERE LOGIN_NAME=\''.$RF_ID.'\'';
 		
-		$station_sql='SELECT DEVICE_NAME FROM DSSC2.ADM_DEV WHERE LOGIN_NAME='.$data['RF_ID'];
-		$res = $conn->query($station_sql);
-		$station_info = $res->fetch(PDO::FETCH_ASSOC);
+		$stid = oci_parse($conn, $station_sql);
+		oci_define_by_name($stid, 'DEVICE_NAME', $device_name);
+		oci_execute($stid);
+		oci_fetch($stid);
+		
+		if(!$device_name){
+			return array('success'=>false,'message'=>'addRfidDataToNingbo failed,station not existed!');
+		}
 		
-		$RF_ID=strtoupper($data['mac']);
 		foreach($data['labels'] as $val){
 			$RF_STAT=0;
+			$plate_no='';
 			if($val['event']['entry']==1){
 				$RF_STAT=1;
 			}elseif($val['event']['leave']==1){
 				$RF_STAT=2;
 			}
 			$RF_FLAGID=strtoupper($val['id']);
+			$vehicle_sql='SELECT o.PLATE_NO FROM DSSC3.W_DW_NON_MOTOR o,DSSC3.W_DW_RFID_TAGS s WHERE s.RFID_SN =\''.$RF_FLAGID.'\'  AND o.rfid_id = s.id  ';
+		
+			$stid = oci_parse($conn, $vehicle_sql);
+			oci_define_by_name($stid, 'PLATE_NO', $plate_no);
+			oci_execute($stid);
+			oci_fetch($stid);
+			if(!$plate_no){
+				continue;
+			}
 			$RF_DATE=date('Y-m-d H:i:s',$val['time']);
 			$handle_data=array(
 				'RF_STAT'=>$RF_STAT,
 				'RF_FLAGID'=>$RF_FLAGID,
 				'RF_ID'=>$RF_ID,
 				'time'=>$val['time'],
-				'address'=>$station_info['DEVICE_NAME']
+				'address'=>$device_name,
+				'plate_no'=>$plate_no
 			);
 			
 			
@@ -56,15 +76,14 @@ class RouteRfidKafkaAction extends Action {
 			//违规行驶检测  超速逆行检测
 			$this->checkIllegalDriving($handle_data,$conn);
 			$sql = 'INSERT INTO "DSSC2"."W_DW_RF_RECORD"("ID", "RF_ID", "RF_FLAGID", "RF_DATE", "RF_STAT") VALUES (DSSC2.SEQ_W_DW_RF_RECORD.nextval, \''.$RF_ID.'\', \''.$RF_FLAGID.'\', TO_DATE(\''.$RF_DATE.'\', \'SYYYY-MM-DD HH24:MI:SS\'), \''.$RF_STAT.'\')';
-		 	var_dump($sql); 
+		 	//var_dump($sql); 
 			//插入数据到oracle轨迹表
 		 	//$res = $conn -> query($sql);
 		
 		}
 		
-		if ($conn){
-		  $conn = null;
-		}
+		oci_free_statement($stid);
+		oci_close($conn);
 		
 		 return array('success'=>true,'message'=>'add success');
 	}
@@ -81,7 +100,7 @@ class RouteRfidKafkaAction extends Action {
 		if (empty($group)) {
 			exit("ROUTE_INDEX_KAFKA_GROUP must be config!".PHP_EOL);
 		}
-		$topics = 'ningbo_dahua';
+		$topics = C('ROUTE_INDEX_KAFKA_TOPIC');
 		if (empty($topics)) {
 			exit("ROUTE_INDEX_KAFKA_TOPIC must be config!".PHP_EOL);
 		}
@@ -133,8 +152,8 @@ class RouteRfidKafkaAction extends Action {
 		        case RD_KAFKA_RESP_ERR_NO_ERROR:
 						 $data = json_decode($message->payload,true);
 						 if( $data ){
-							 //$this->addRfidDataToNingbo($data);
-							 $this->addRfidDataToRenlian($data);
+							 $this->addRfidDataToNingbo($data);
+							 //$this->addRfidDataToRenlian($data);
 						 }
 		            break;
 		        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
@@ -163,10 +182,9 @@ class RouteRfidKafkaAction extends Action {
 		}
 		
 		
-		static $recvCount = 0;
-		$recvCount++;
-		var_dump($recvCount . ".".json_encode($data));
-		return;
+		
+		var_dump($data);
+		
 		
 		
 		$conn = null;
@@ -204,91 +222,91 @@ class RouteRfidKafkaAction extends Action {
 	}
 	
  
-	private  function addControlAlarm( $data, $conn ){
-		
-		$vehicle_sql='SELECT o.PLATE_NO,s.RFID_SN FROM DSSC3.W_DW_NON_MOTOR o,DSSC3.W_DW_RFID_TAGS s WHERE s.RFID_SN ='.$data['RF_FLAGID'].'  AND o.rfid_id = s.id  ';
-		$res2 = $conn->query($vehicle_sql);
-		$vehicle_info = $res2->fetch(PDO::FETCH_ASSOC);
-		if(!$data['address']){
-			$station_sql='SELECT DEVICE_NAME FROM DSSC2.ADM_DEV WHERE LOGIN_NAME='.$data['RF_ID'];
-			$res = $conn->query($station_sql);
-			$station_info = $res->fetch(PDO::FETCH_ASSOC);
-			$data['address']=$station_info['DEVICE_NAME'];
-		}
+	private  function addControlAlarm( $data ){
+		
 		$save_data=array(
-						'plate_no'=>$vehicle_info['PLATE_NO'],
+						'plate_no'=>$data['plate_no'],
 						'rfid_sn'=>$data['RF_FLAGID'],
 						'address'=>$data['address'],
-						'alarm_type'=>$data['alarm_typ'],
+						'alarm_type'=>$data['alarm_type'],
 						'created_at'=>$data['time'],
-						'remark'=>$data['remark']
+						'remark'=>$data['remark'],
+						'state'=>1,
 		);
 		$res=M('control_alarm')->createAdd($save_data);
 		return $res;
 	}
 	
  
-	private  function checkControlAlarm( $data, $conn ){
+	private  function checkControlAlarm( $data ){
 		//先检测标签是否布控
-			$cond=array('control_obj'=>$data['RF_FLAGID']);
+			$cond=array('control_obj'=>array('in',[$data['RF_FLAGID'],$data['plate_no']])  );
 			$ve_con=M('control_manage')->where($cond)->find();
+		
 			//存在布控标签 并在时间内
-			if($ve_con && ($val['time']>$ve_con['start_time']) && ($val['time']<$ve_con['end_time'])){
-				$this->addControlAlarm($data,$conn);
+			if($ve_con && ($data['time']>$ve_con['start_time']) && ($data['time']<$ve_con['end_time'])){
+				$data['alarm_type']='control';
+				$this->addControlAlarm($data);
 			}
+		
 			//检测区域布控
 			$cond2=array('control_obj'=>array("LIKE", '%'.$data['RF_ID'].'%'));
 			$sta_con=M('control_manage')->where($cond2)->find();
 			//存在布控基站
-			if($sta_con && ($val['time']>$sta_con['start_time']) && ($val['time']<$sta_con['end_time'])){
-				if($sta_con['bw_ids']){
+			if($sta_con && ($data['time']>$sta_con['start_time']) && ($data['time']<$sta_con['end_time'])){
+				if($sta_con['bw_ids']){//存在名单
 					
 					$bwIdArr=explode(',',$sta_con['bw_ids']);
-					$info=M('bw_list')->where( array('id'=>array('in',$bwIdArr) ) )->find();
-					//判断是黑名单还是白名单
+					//获取名单内标签
 					$rfid_arr=M('rfid_with_bw')->where(array('bw_id'=>array('in',$bwIdArr) ))->getField('rfid',true);
-					if($sta_con['area_type']=='1'){
-						//禁止活动区域 
-						if($info['type']=='0'){
-							//指定黑名单禁止 
+					
+					if($sta_con['area_type']=='1'){//禁止活动区域
+						
+						if($sta_con['bw_type']=='0'){//指定黑名单禁止 
+						
 							if(in_array($data['RF_FLAGID'],$rfid_arr)){
 								//在黑名单内 告警
-								$data['alarm_typ']='forbid_b';
-								$this->addControlAlarm($data,$conn);
+								$data['alarm_type']='forbid_in';
+								$data['remark']='驶入黑名单禁入区域';
+								$this->addControlAlarm($data);
 							}
 						}else{
 							//白名单的不禁止   
 							if(!in_array($data['RF_FLAGID'],$rfid_arr)){
 								//不在白名单内 告警
-								$data['alarm_typ']='forbid_w';
-								$this->addControlAlarm($data,$conn);
+								$data['alarm_type']='forbid_in';
+								$data['remark']='驶入禁入区域';
+								$this->addControlAlarm($data);
 							}
 						}
 						
-					}else{
-						//活动区域
+					}else{//活动区域
+						
 						//指定黑名单
 						if(in_array($data['RF_FLAGID'],$rfid_arr)){
 								//在黑名单内 告警
-							$data['alarm_typ']='activity';
-							$this->addControlAlarm($data,$conn);
+							$data['remark']='驶入活动区域';
+							$data['alarm_type']='activity_in';
+							$this->addControlAlarm($data);
 						}
 					}
 				}else{
 					//无黑白名单布控  全部禁止
-					$data['alarm_typ']='forbid_all';
-					$this->addControlAlarm($data,$conn);
+					$data['remark']='驶入禁入区域';
+					$data['alarm_type']='forbid_in';
+					$this->addControlAlarm($data);
 				}
 			}
 	}
 	
  
-	private  function checkIllegalDriving( $data, $conn ){
+	private  function checkIllegalDriving( $data ){
 		$redis = Redis("nbfd_stuck_section_data","hash");
 		
 		//先查询基站是否设置卡点
 		$cond=array('macs'=>array("LIKE", '%'.$data['RF_ID'].'%'));
 		$section_id=M('stuck_point')->where($cond)->getField('id');
+		
 		//检测是否是前置卡点
 		$pre_section_cond=array('pre_spot'=>$section_id);
 		$pre_section=M('stuck_section')->where($pre_section_cond)->find();
@@ -297,10 +315,9 @@ class RouteRfidKafkaAction extends Action {
 			$key = "stuck_section_".$pre_section['id']."_".$data['RF_FLAGID'];
 			$passInfo=json_decode($redis->get($key),true);
 			//存在逆行进入卡点时间  且启用超速检测
-			if($passInfo && ($passInfo['section']=='pos') && ($pos_section['retrograde_stat']==1)){
-				$data['alarm_typ']='retrograde';
-				
-				$this->addControlAlarm($data,$conn);
+			if($passInfo && ($passInfo['section']=='pos') && ($pre_section['retrograde_stat']=='1')){
+				$data['alarm_type']='retrograde';
+				$this->addControlAlarm($data);
 			}
 			$redisData = array(
 				$key =>json_encode(array(
@@ -315,19 +332,26 @@ class RouteRfidKafkaAction extends Action {
 		//检查是否是后置卡点
 		$pos_section_cond=array('pos_spot'=>$section_id);
 		$pos_section=M('stuck_section')->where($pos_section_cond)->find();
+		
 		if($pos_section){
 			//后置卡点 取标签进入卡点区间时间
 			$key = "stuck_section_".$pos_section['id']."_".$data['RF_FLAGID'];
 			$passInfo=json_decode($redis->get($key),true);
+		
 			//存在进入卡点时间  且启用超速检测
-			if($passInfo && ($passInfo['section']=='pre') && ($pos_section['over_speed_stat']==1)){
+			if($passInfo && ($passInfo['section']=='pre') && ($pos_section['over_speed_stat']=='1')){
+				var_dump($data['time']);
+				var_dump($passInfo['time']);
 				$hour= ($data['time']-$passInfo['time'])/3600;
+				var_dump($hour);
 				$speed=($pos_section['distance']/1000)/$hour;
-				if($speed>$max_speed){
+				var_dump($speed);
+				var_dump($pos_section['max_speed']);
+				if($speed>$pos_section['max_speed']){
 					//超速行驶
-					$data['alarm_typ']='over_speed';
+					$data['alarm_type']='over_speed';
 					$data['remark']='速度：'. round($speed,2);
-					$this->addControlAlarm($data,$conn);
+					$this->addControlAlarm($data);
 				}
 			}
 			//存经过后置卡点时间
@@ -360,7 +384,7 @@ class RouteRfidKafkaAction extends Action {
 		//统计表数据添加
 		$to_cond=array(
 			'mac'=>$data['RF_ID'],
-			'date'=>date('Y-m-d',$data['time']);
+			'date'=>date('Y-m-d',$data['time'])
 		);
 		if(!M('station_passing')->where($to_cond)->count()){
 			$total_data=array(
@@ -376,5 +400,109 @@ class RouteRfidKafkaAction extends Action {
 		return $res;
 	}
 	
+ 
+	public  function test(  ){
+		$topic = C('ROUTE_INDEX_KAFKA_TOPIC');
+		
+		
+				static $rk;
+				if (!extension_loaded('rdkafka')){
+					echo 'pushToKafka fail,extension of rdkafka has not installed!!'.PHP_EOL;
+					return false;
+				}
+				if(!$rk){
+					$conf = new Rdkafka\Conf();
+					$conf->set('batch.num.messages', 2);
+					//$conf->set('linger.ms', 10);
+					//$conf->set('log_level', (string) LOG_DEBUG);
+					//$conf->set('debug', 'all');
+					$conf->setErrorCb(function($producer, $msg) {
+						printf("%s: %s\n", rd_kafka_err2str($err), $errstr);
+					});
+					$conf->setDrMsgCb(function($producer, $msg) {
+						if($msg->err) {
+							echo 'Message delivery failed:' . $msg->errstr();
+						} else {
+							echo "sent message sucessfully.";
+						}
+					});
+					$rk = new RdKafka\Producer($conf);
+				}
+				var_dump($topic);
+				//var_dump(C('KAFKA_BROKER_LIST'));die;
+				//$rk->setLogLevel(LOG_DEBUG);
+				$rk->addBrokers(C('KAFKA_BROKER_LIST'));
+				$topic = $rk->newTopic($topic);
+				$res='{"methond":"track","mac":"FF04B5C4","gps":{"locationState":"A","lat":0,"latType":"N","lng":0,"lngType":"E"},"labels":[{"id":"0308EC58","event":{"dec":8,"lowBattery":0,"entry":0,"leave":1,"in":0},"time":"'.time().'"},{"id":"01000424","event":{"dec":8,"lowBattery":0,"entry":0,"leave":1,"in":0},"time":"1667529922"},{"id":"01000422","event":{"dec":20,"lowBattery":0,"entry":1,"leave":0,"in":1},"time":"1667529922"}]}';
+							
+		
+				$topic->produce(RD_KAFKA_PARTITION_UA, 0,$res);	
+				
+				
+				$rk->poll(0);
+				while ($rk->getOutQLen() > 0) {
+					$rk->poll(1);
+				}
+				
+			
+	}
+	
+ 
+	public  function test2(  ){
+		$topic = C('ROUTE_INDEX_KAFKA_TOPIC');
+		
+		
+				static $rk;
+				if (!extension_loaded('rdkafka')){
+					echo 'pushToKafka fail,extension of rdkafka has not installed!!'.PHP_EOL;
+					return false;
+				}
+				if(!$rk){
+					$conf = new Rdkafka\Conf();
+					$conf->set('batch.num.messages', 2);
+					//$conf->set('linger.ms', 10);
+					//$conf->set('log_level', (string) LOG_DEBUG);
+					//$conf->set('debug', 'all');
+					$conf->setErrorCb(function($producer, $msg) {
+						printf("%s: %s\n", rd_kafka_err2str($err), $errstr);
+					});
+					$conf->setDrMsgCb(function($producer, $msg) {
+						if($msg->err) {
+							echo 'Message delivery failed:' . $msg->errstr();
+						} else {
+							echo "sent message sucessfully.";
+						}
+					});
+					$rk = new RdKafka\Producer($conf);
+				}
+				var_dump($topic);
+				//var_dump(C('KAFKA_BROKER_LIST'));die;
+				//$rk->setLogLevel(LOG_DEBUG);
+				$rk->addBrokers(C('KAFKA_BROKER_LIST'));
+				$topic = $rk->newTopic($topic);
+				$res='{"methond":"track","mac":"FF04C526","gps":{"locationState":"A","lat":0,"latType":"N","lng":0,"lngType":"E"},"labels":[{"id":"0308EC58","event":{"dec":8,"lowBattery":0,"entry":0,"leave":1,"in":0},"time":"'.time().'"},{"id":"01000424","event":{"dec":8,"lowBattery":0,"entry":0,"leave":1,"in":0},"time":"1667529922"},{"id":"01000422","event":{"dec":20,"lowBattery":0,"entry":1,"leave":0,"in":1},"time":"1667529922"}]}';
+							$info = array(
+					'DeviceId' => '869688888888888',
+					//'State' => (string)$data['state'],
+					//'Speed' => $data['speed'],
+					 'Latitude'=>30.192289977186,
+		          'Longitude'=>120.20063757299,
+					'DeviceTime' => time(),
+					'Altitude'=>108.951,
+					//'LBS' => $data['lbs'],
+					//'Direction' => $data['direction'],
+				);
+		
+				$topic->produce(RD_KAFKA_PARTITION_UA, 0,$res);	
+				
+				
+				$rk->poll(0);
+				while ($rk->getOutQLen() > 0) {
+					$rk->poll(1);
+				}
+				
+			
+	}
+	
 
 }