czapp-api/Home/Lib/Action/CronAction.class.php

<?php



class CronAction extends Action {
			


 
	public  function createEplate_index(  ){
		$this->createElectronicPlate();
	}
	
 
	private  function createElectronicPlate(  ){
		$start = time();
		$pendingElectricPlate = Redis('jyzl_wait_create_eplate','queue');
		$plateLocalPath = Redis('jyzl_wait_upload2oss_eplate', 'queue');
		while( (time() - $start) < 60 ){
			$licensPlate = $pendingElectricPlate->pop();
			if(!$licensPlate){
				echo 'no message!'.PHP_EOL;
				sleep(1);
				continue;
			}
			echo 'pop licensPlate:'.$licensPlate.PHP_EOL;
			$field = 'LicensePlate,VehicleColor, FullName, DetailedAdd, FrameNumber, MotorNumber, VehicleBrand, RegistrationTime';
			$vehicleInfo = M('jms_vehicle')->where(array('LicensePlate' => $licensPlate))->field($field)->find();
			if(!$vehicleInfo){
				echo 'vehicleInfo not existed,$licensPlate = '.$licensPlate .PHP_EOL;
				continue;
			}
			//生成电子车牌到本地
			$localPath = $this->createLocalElectronicPlate($vehicleInfo);
			if(!$localPath){
				echo 'createLocalElectronicPlate failed,$licensPlate = '.$licensPlate .PHP_EOL;
				//生成失败的重新放回队列
				$result = $pendingElectricPlate->add($licensPlate);
				if(!$result){
					echo 'pendingElectricPlate->add() failed,$licensPlate = '.$licensPlate .PHP_EOL;
				}
				continue;
			}
			//生成的本地电子车牌，加入到待上传oss队列
			$up2ossWait = json_encode(array('licensePlate' => $licensPlate, 'localPath' => $localPath));
			$result = $plateLocalPath->add($up2ossWait);//push方法没有返回值，用add代替
			if(!$result){
				echo 'plateLocalPath->add() failed,$licensPlate = '.$licensPlate .PHP_EOL;
				continue;
			}
			usleep(100000);
		}
	}
	
 
	private  function uploadElectronicPlate2Oss(  ){
		$config = array(
			'OssDsn' => C('OSS_DSN'),
			"SaveRule" => "/electronic_plate/{Y}{m}{d}/{uid}.{ext}",
			"AllowExts" => array('jpg', 'png', 'jpeg'), // 允许上传的文件后缀（留空为不限制）
			"ResizeImage" => false,   // 是否自动压缩
			"MaxImageWidth" => 1024,
			"MaxImageHeight" => 1024,
			"IsCheckRgb" => false,
			"MinImgAverageRgb" => 70
		);
		
		$backImgLocalPath = ENTRY_PATH .'/Public/images/back.jpg';
		$upload = new \Jms\File\Oss2($config);
		//电子车牌背部固定图片第一次上传至oss
		$plateBackOssUrl = S('czapp_cache_plate_oss_url');
		if(!$plateBackOssUrl){
			$uploadBack = $upload->localFileUpload($backImgLocalPath);
			if(!$uploadBack['success']){
				echo $uploadBack['message'].PHP_EOL;
				exit;
			}else{
				$plateBackOssUrl = $uploadBack['objectname'];
				if(!$plateBackOssUrl){
					echo 'upload backImg success! But cant get the ossUrl';
					exit;
				}
				S('czapp_cache_plate_oss_url', $plateBackOssUrl);
			}
		}
		$start = time();
		$plateLocalPath = Redis('jyzl_wait_upload2oss_eplate', 'queue');
		while( (time() - $start) < 60 ){	
			$data = $plateLocalPath->pop();
			if(!$data){
				echo 'no more data, waiting for next mesaage.'.PHP_EOL;
				sleep(1);
				continue;
			}
		
			$fileName = $data['localPath'];
			$licensPlate = $data['licensePlate'];
			if(!$licensPlate || !$fileName){
				echo 'cant get right licensPlate or fileName'.PHP_EOL;
				echo 'fileName:'.$fileName.PHP_EOL;
				echo 'licensPlate:'.$licensPlate.PHP_EOL;
				echo '---------------delimiter----------------'.PHP_EOL;
				continue;
			}
			//上传至oss
			$uploadRes = $upload->localFileUpload($fileName);
			//失败重回队列
			if(!$uploadRes['success']){
				echo $uploadRes['message'].PHP_EOL;
				$result = $plateLocalPath->add($data);
				if(!$result){
					echo 'plateLocalPath->add() failed,$licensPlate = '. $data['licensPlate'] .PHP_EOL;
					continue;
				}
			}else{
				echo 'upload plateFrontImg success!'.PHP_EOL;
			}
			
			$frontImageUrl = $uploadRes['objectname'];
			if(!$frontImageUrl){
				echo 'cant get img_oss_url'.PHP_EOL;
				continue;
			}
			echo 'return oss-url:'.$frontImageUrl.PHP_EOL;
		
		//	$updateRes = M('jms_vehicle')->where(array('LicensePlate' => $licensPlate))->setField('FrontElectronicPlateUrl',$frontImageUrl);
			$saveData = array( 'FrontElectronicPlateUrl' => $frontImageUrl, 'BackElectronicPlateUrl' => $plateBackOssUrl);
			$updateRes = M('jms_vehicle')->where(array('LicensePlate' => $licensPlate))->save($saveData);
			
			if(!$updateRes){
				echo 'save electricPlate in mysql failed'.PHP_EOL;
				continue;
			}
			echo 'save electricPlate in mysql success'.PHP_EOL;
			
			//删除本地图片
			$filePath = $filename;
			if(!file_exists($filePath)){
				$flag = unlink($filePath);
				if(!$flag){
					echo 'delete localImg failed'.PHP_EOL;
				}else{
					echo 'delete localImg success'.PHP_EOL;
				}
			}
			usleep(100000);
		}
		
	}
	
 
	public  function uploadEplate_index(  ){
		$this->uploadElectronicPlate2Oss();
	}
	
 
	public  function test_function(  ){
		$vehicleInfo = array(
			'LicensePlate' => 'LY100021',
			'VehicleColor' => '白色',
			'FullName' => '张三', 
			'Address' => '浙江杭州滨江195号',
			'FrameNumber' => '123456789',
			'MotorNumber' => '233456789', 
			'VehicleBrand'=> '雅迪z123', 
			'RegistrationTime' => '2019-05-15 12:12:00'
		);
		$localPath = $this->createLocalElectronicPlate($vehicleInfo);
		echo $localPath .PHP_EOL;
		exit;
		
		$plate = 'BJ000100';
		$pendingElectricPlate = Redis("jyzl_wait_create_eplate","queue");
		$licensePlate = $plate;
		$pendingElectricPlate->push($licensePlate);
		
	}
	
 
	public  function acrossAlarm2Kafka(  ){
		// 从 topic ：gps_location_data 取轨迹
		$conf = new RdKafka\Conf();
		// Set a rebalance callback to log partition assignments (optional)（当有新的消费进程加入或者退出消费组时，kafka 会自动重新分配分区给消费者进程，这里注册了一个回调函数，当分区被重新分配时触发）
		$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
		    switch ($err) {
		        case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
		            echo "Assign: ";
		            var_dump($partitions);
		            $kafka->assign($partitions);
		            break;
		
		         case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
		             echo "Revoke: ";
		             var_dump($partitions);
		             $kafka->assign(NULL);
		             break;
		
		         default:
		            throw new \Exception($err);
		    }
		});
		// Configure the group.id. All consumer with the same group.id will consume（ 配置groud.id 具有相同 group.id 的consumer 将会处理不同分区的消息，所以同一个组内的消费者数量如果订阅了一个topic， 那么消费者进程的数量多于这个topic 分区的数量是没有意义的。）
		// different partitions.
		$conf->set('group.id', 'fenceAlarmMsgGroup');
		if( !C('KAFKA_BROKER_LIST') ){
			echo 'please set broker list !!! ';
		}
		// Initial list of Kafka brokers（添加 kafka集群服务器地址）
		$conf->set('metadata.broker.list', C('KAFKA_BROKER_LIST'));
		$topicConf = new RdKafka\TopicConf();
		// Set where to start consuming messages when there is no initial offset in
		// offset store or the desired offset is out of range.
		// 'smallest': start from the beginning
		$topicConf->set('auto.offset.reset', 'smallest');
		// Set the configuration to use for subscribed/assigned topics
		$conf->setDefaultTopicConf($topicConf);
		$consumer = new RdKafka\KafkaConsumer($conf);
		// 订阅轨迹数据topic
		$consumer->subscribe(['gps_location_data']);
		while (true) {
			$message = $consumer->consume(120*1000);
			switch ($message->err) {
				case RD_KAFKA_RESP_ERR_NO_ERROR:
				// 判断是否超出围栏范围 ，存入 topic：gps_alarm_msg_queue 
				$route_info = json_decode($message->payload,true);
				if( empty($route_info) ){
					echo 'empty route info.';
					break;
				}
				$result = $this->produceAcrossAlarmData($route_info);
				if($result){
					echo $result,PHP_EOL;
					debug_log('across_alarm',$result['message']);
				}
				break;
				case RD_KAFKA_RESP_ERR__PARTITION_EOF:
				echo "No more messages; will wait for more\n";
				break;
				case RD_KAFKA_RESP_ERR__TIMED_OUT:
				echo "Timed out\n";
				break;
				default:
				throw new \Exception($message->errstr(), $message->err);
				break;
			}
		}
	}
	
 
	private  function produceAcrossAlarmData( $route_info ){
		if( !$route_info['DeviceId'] ){
			return array('success' => false, 'message' => 'device id is not exists.');
		}
		if( !$route_info['Longitude'] ){
			return array('success' => false, 'message' => 'longitude is not exists.');
		}
		if( !$route_info['Latitude'] ){
			return array('success' => false, 'message' => 'latitude is not exists.');
		}
		if( !$route_info['DeviceTime'] ){
			return array('success' => false, 'message' => 'device time is not exists.');
		}
		// 从数据库中取出车牌号，缓存1天
		if( S('plate-'.$route_info['DeviceId']) ){
			$plate = S('plate-'.$route_info['DeviceId']);
		}else{
			$where = array('GpsDeviceNumber'=>$route_info['DeviceId']);
			$plate = M('jms_vehicle')->where($where)->getField('LicensePlate');
			S('plate-'.$route_info['DeviceId'], $plate, 24*60*60);
		}
		// 是否启用围栏
		$rlfd_vehicle_fence = Redis('rlfd_vehicle_fence','hash');
		$fence = $rlfd_vehicle_fence->get($plate);
		$fence = json_decode($fence, true);
		
		if( empty($fence) ){
			return array('success' => false, 'message' => '围栏信息不存在');
		}
		if( !$fence['fenceStatus'] ){
			return array('success' => false, 'message' => '未启用围栏');
		}
		$fence_info = $fence['fenceInfo'];
		if( empty($fence_info['data']) ){
			return array('success' => false, 'message' => '围栏坐标数据不存在');
		}
		// 是否越界
		$route_point = array(
			'lng' => $route_info['Longitude'],
			'lat' => $route_info['Latitude']
		);
		$result = true; // 默认在围栏内
		if( $fence_info['type'] == 'circle' ){ // 圆形围栏
			$distance = \Jms\Algo\Geometry::distanceBetween2BdPoints($fence_info['data']['center'], $route_point); //获取轨迹点到围栏中心点间距离,km
			$result = $distance*1000 < $fence_info['data']['radius'];// 距离圆心大于半径说明越界了
		}elseif( $fence_info['type'] == 'polygon' ){ // 多边形围栏
			$result = \Jms\Algo\Geometry::isInPolygon($fence_info['data']['vertex'], $route_point);
		}else{
			return array('success' => false, 'message' => '未知围栏类型');
		}
		if( !$result ){
			$alarm_data = array(
			  //"type" => C('FENCE_ALARM'),
			  "type" => \Rlfd\Alarm\PushTypeEnum::FENCE_ALARM,
			  "title" => "超出电子围栏",
			  "content" => "车辆 {$plate} 已超出设置的电子围栏范围，请前往停车处确认是否被盗。",
			  "device_number" => $route_info['DeviceId']
			);
			kafkaProducer('gps_alarm_msg_queue', $alarm_data); // 添加到kafka
			return array('success' => true, 'message' => '添加围栏告警消息到 gps_alarm_msg_queue');
		}
		return array('success' => false, 'message' => '没有超出围栏');
		
	}
	
 
	public  function mockProduce(  ){
		/* // jyzl gps
		$msg_data = array(
			'DeviceId' => FFFFFF123122,
			'State' => 1,
			'Speed' => 1.2,
			'Longitude' => 121.20638,
			'Latitude' => 30.18852,
			'DeviceTime' => date('Y-m-d H:i:s'),
			'LBS' => 'LBS',
			'Direction' => 's',
		);
		kafkaProducer('gps_location_data',$msg_data);*/
		
		// fly 轨迹数据
		$msg_data = array(
			'StationCode' => '30B5F1012539',
			"Longitude" => "120.600889",
			"Latitude" => "30.191478",
			"Address" => "华城·和瑞科技广场(长河路475号)",
			'StationType' => 0,
			"CityId" => "2902",
			"StationName" => "基站名称",
			"AddTime" => date('Y-m-d H:i:s',strtotime('-3 minutes')),
			"VehicleNumber" => "800000241B",
			"VehicleStatus" =>  "1",
			"SignalCount" => 88,
			"OnlineTime" => date('Y-m-d H:i:s',strtotime('-2 minutes'))
		);
		kafkaProducer('fly_vroute_data',$msg_data);
	}
	
 
	private  function createLocalElectronicPlate( $params ){
		$license_plate = $params['LicensePlate'];
		if(!$license_plate){
			echo "LicensePlate empty!".PHP_EOL;
			return false;
		}
		$vehicle_color = $params['VehicleColor'];
		if(!$vehicle_color){
			echo "VehicleColor empty!".PHP_EOL;
			return false;
		}
		$real_name = $params['FullName'];
		if(!$real_name){
			echo "FullName empty!".PHP_EOL;
			return false;
		}
		$address = $params['DetailedAdd'];
		if(!$address){
			echo "DetailedAdd empty!".PHP_EOL;
			return false;
		}
		
		$cjh = $params['FrameNumber'];
		$djh = $params['MotorNumber'];
		/*
		if(!$cjh){
			echo "FrameNumber empty!".PHP_EOL;
			return false;
		}
		
		if(!$djh){
			echo "MotorNumber empty!".PHP_EOL;
			return false;
		}
		*/
		$cph = $params['VehicleBrand'];
		if(!$cph){
			echo "VehicleBrand empty!".PHP_EOL;
			return false;
		}
		$regist_time = strtotime($params['RegistrationTime']);
		if($regist_time < 1546272000){
			echo "RegistrationTime invalid! RegistrationTime: ".$params['RegistrationTime'].PHP_EOL;
			return false;
		}
		$date = date('Y-m-d',$regist_time);
		if(!$date){
			echo "date empty! RegistrationTime: ".$params['RegistrationTime'].PHP_EOL;
			return false;
		}
		$reg_date = $date;
		$fz_date = $date;
		$fz_org = "包头市公安局";
		
		$im = imagecreatetruecolor(500, 278);           // 设置画布
		//$bg = imagecreatefromjpeg('bg.jpg');   // 设置背景图片
		$front_img = ENTRY_PATH . '/Public/images/front.jpg';
		if(!is_file($front_img)){
			echo "front_img not existed! front_img: ".$front_img.PHP_EOL;
			return false;
		}
		$bg = imagecreatefromjpeg($front_img);   // 设置背景图片
		imagecopy($im,$bg,0,0,0,0,500,278);             // 将背景图片拷贝到画布相应位置
		imagedestroy($bg);                              // 销毁背景图片
		$font = ENTRY_PATH .'/Public/font/stsong.ttf';               // 设置字体             // 设置字体，这里可以指向ttf文件
		if(!is_file($font)){
			echo "font file not existed! font: ".$font.PHP_EOL;
			return false;
		}
		$blacka = imagecolorallocate($im, 15, 23, 25);  // 颜色
		
		/* 写入内容 */
		imagettftext($im, 12, 0, 135, 66, $blacka, $font,$license_plate );       // 车牌号
		imagettftext($im, 12, 0, 335, 66, $blacka, $font,$vehicle_color );       // 车辆颜色
		imagettftext($im, 12, 0, 135, 96, $blacka, $font,$real_name );       // 姓名
		imagettftext($im, 12, 0, 135, 128, $blacka, $font,$address );       // 住址
		imagettftext($im, 12, 0, 135, 160, $blacka, $font,$cjh );       // 车架号
		imagettftext($im, 12, 0, 335, 160, $blacka, $font,$djh );       // 电机号
		imagettftext($im, 12, 0, 263, 192, $blacka, $font,$cph );       // 厂牌型号
		imagettftext($im, 11, 0, 260, 222, $blacka, $font,$reg_date );       // 注册日期
		imagettftext($im, 11, 0, 376, 222, $blacka, $font,$fz_date );       // 发证期
		imagettftext($im, 12, 0, 263, 255, $blacka, $font,$fz_org );       // 发证机关
		
		
		$img_file_dir = SOLUTION_LOG_PATH ."/images/".date('Y-m-d')."/";
		if(!is_dir($img_file_dir)){
			$res = mkdir($img_file_dir,0777,true);
			if (!$res){
				echo "目录 $img_file_dir 创建失败！".PHP_EOL;
				return false;
			}
		}
		$img_file = $img_file_dir .$license_plate.".jpg";
		$result = imagejpeg($im, $img_file);     // 生成jpeg格式图片
		imagedestroy($im);  // 销毁图片
		if(!$result){
			echo "生成电子车牌失败, license_plate: ".$license_plate .PHP_EOL;
			return false;
		}
		
		echo "生成电子车牌完成, license_plate: ".$license_plate .PHP_EOL;
		return $img_file;
	}
	
 
	private  function produceLockAlarmMessage( $route_info ){
		if( !$route_info['DeviceId'] ){
			return array('success' => false, 'message' => 'device id is not exists.');
		}
		if( !$route_info['Longitude'] ){
			return array('success' => false, 'message' => 'longitude is not exists.');
		}
		if( !$route_info['Latitude'] ){
			return array('success' => false, 'message' => 'latitude is not exists.');
		}
		if( !$route_info['DeviceTime'] ){
			return array('success' => false, 'message' => 'device time is not exists.');
		}
		
		// 从数据库中取出车牌号，缓存1天
		if( S('plate-'.$route_info['DeviceId']) ){
			$plate = S('plate-'.$route_info['DeviceId']);
		}else{
			$where = array('GpsDeviceNumber'=>$route_info['DeviceId']);
			$plate = M('jms_vehicle')->where($where)->getField('LicensePlate');
			S('plate-'.$route_info['DeviceId'], $plate, 24*60*60);
		}
		
		
		
		//redis获取锁车状态
		$vehicle = Redis('czapp_lock_status','hash');
		$lockStatus = $vehicle->get($plate);
		if(!$lockStatus){
			return array('success' => false, 'message' => $plate.'-'.$route_info['DeviceId'].' 未启用锁车');
		}
		
		$lockInfoRedis = Redis('czapp_lockinfo', 'hash');
		$lockInfoRes = $lockInfoRedis->get($route_info['DeviceId']);
		if(!$lockInfoRes){
			return array('success' => false, 'message' => $plate.'-'.$route_info['DeviceId'].' 无法查到redis锁车信息');
		}
		$lockInfoRes = json_decode($lockInfoRes, true);
		
		/*
		$deviceSleepTime = C('GPS_SLEEP_TIME');//gps休眠时间
		if(!$deviceSleepTime){
			$deviceSleepTime = 300;
		}
		$timeInterval = time() - strtotime($lockInfoRes['GpsOnlineTime']);
		if($timeInterval < $deviceSleepTime){//小于休眠时间不报警
				echo '--3'.PHP_EOL;
			return array('success' => false, 'message' => $plate.'-'.$route_info['DeviceId'].' 小于休眠时间不告警');
		}
		*/
		
		//是否移动
		$distance = \Jms\Algo\Geometry::getBdDistance($lockInfoRes['GpsLatitude'], $lockInfoRes['GpsLongitude'], $route_info['Latitude'], $route_info['Longitude']) * 1000;
		
		$configDist = C('LOCK_ALARM_DISTANCE');//锁车后移动距离
		if(!$configDist){
			$configDist = 20;
		}
		
		$alarm = false;//默认不告警
		if($distance > $configDist){
			$alarm = true;
		}
		
		
		if($alarm){
			$alarm_data = array(
				//"type" => C('FENCE_ALARM'),
				"type" => \Rlfd\Alarm\PushTypeEnum::LOCK_VEHICLE_ALARM,
				"title" => "锁车告警",
				"content" => "车辆 {$plate} 有异常移动，请前往停车处确认是否被盗。",
				"device_number" => $route_info['DeviceId']
			);
			kafkaProducer('gps_alarm_msg_queue', $alarm_data); // 添加到kafka
			return array('success' => true, 'message' => $plate.'-'.$route_info['DeviceId'].' 添加告警消息到 gps_alarm_msg_queue');
		}
		return array('success' => false, 'message' => $plate.'-'.$route_info['DeviceId'].' 无异常');
		
	}
	
 
	public  function lockAlarm2Kafka(  ){
		// 从 topic ：gps_location_data 取轨迹
		$conf = new RdKafka\Conf();
		// Set a rebalance callback to log partition assignments (optional)（当有新的消费进程加入或者退出消费组时，kafka 会自动重新分配分区给消费者进程，这里注册了一个回调函数，当分区被重新分配时触发）
		$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
		    switch ($err) {
		        case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
		            echo "Assign: ";
		            var_dump($partitions);
		            $kafka->assign($partitions);
		            break;
		
		         case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
		             echo "Revoke: ";
		             var_dump($partitions);
		             $kafka->assign(NULL);
		             break;
		
		         default:
		            throw new \Exception($err);
		    }
		});
		// Configure the group.id. All consumer with the same group.id will consume（ 配置groud.id 具有相同 group.id 的consumer 将会处理不同分区的消息，所以同一个组内的消费者数量如果订阅了一个topic， 那么消费者进程的数量多于这个topic 分区的数量是没有意义的。）
		// different partitions.
		$conf->set('group.id', 'lockAlarmMsgGroup');
		if( !C('KAFKA_BROKER_LIST') ){
			echo 'please set broker list !!! ';
		}
		// Initial list of Kafka brokers（添加 kafka集群服务器地址）
		$conf->set('metadata.broker.list', C('KAFKA_BROKER_LIST'));
		$topicConf = new RdKafka\TopicConf();
		// Set where to start consuming messages when there is no initial offset in
		// offset store or the desired offset is out of range.
		// 'smallest': start from the beginning
		$topicConf->set('auto.offset.reset', 'smallest');
		// Set the configuration to use for subscribed/assigned topics
		$conf->setDefaultTopicConf($topicConf);
		$consumer = new RdKafka\KafkaConsumer($conf);
		// 订阅轨迹数据topic
		$consumer->subscribe(['gps_location_data']);
		while (true) {
			$message = $consumer->consume(120*1000);
			switch ($message->err) {
				case RD_KAFKA_RESP_ERR_NO_ERROR:
				// 判断是否超出围栏范围 ，存入 topic：gps_alarm_msg_queue 
				$route_info = json_decode($message->payload,true);
				if( empty($route_info) ){
					echo 'empty route info.';
					break;
				}
				$lockAlarmRes = $this->produceLockAlarmMessage($route_info);
				if($lockAlarmRes){
					print_r($lockAlarmRes);
					debug_log('lock_alarm',$lockAlarmRes['message']);
				}
				break;
				case RD_KAFKA_RESP_ERR__PARTITION_EOF:
				echo "No more messages; will wait for more\n";
				break;
				case RD_KAFKA_RESP_ERR__TIMED_OUT:
				echo "Timed out\n";
				break;
				default:
				throw new \Exception($message->errstr(), $message->err);
				break;
			}
		}
	}
	

}