station-test-api/task_script/RFID_STATION_CLIENT.php

<?php
require('../vendor/autoload.php');
use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;
use think\facade\Cache;
// date_default_timezone_set("PRC");
// date_default_timezone_set("America/Bahia");
// define('HOST', 'r-bp1eebab79320044pd.redis.rds.aliyuncs.com');
// define('PORT', '6379');
// define('PASSWORD', '7e2b5c91e438be3c!');
// define('DATABASE', 4);
define('HOST', '127.0.0.1');
define('PORT', '6379');
define('PASSWORD', '123456');
define('DATABASE', 2);

function app_redis()
{
    static $redis = null;
    static $conn = false;
    if (!$conn) {
        connect: //定义标签
        $redis = new Redis();
        try {
            //建立的Redis短连接,在请求结束后不会自动关闭,相当于持久连接.
            $conn = $redis->connect(HOST, PORT);
            $conn = $redis->auth(PASSWORD);
            $conn = $redis->select(DATABASE);
            // 连接成功,返回$redis对象,连接失败,返回false.
            return ($conn === true) ? $redis : false;
        } catch (Exception $e) {
            return false;
        }
    } else {
        // 这里假设PHP-FPM在处理一个请求的时间内,Redis连接都是可用的.
        // 所以只在PHP-CLI下检查Redis连接的状态,进行断线重连.
        if (php_sapi_name() === 'cli') {
            try {
                // ping用于检查当前连接的状态,成功时返回+PONG,失败时抛出一个RedisException对象.
                // ping失败时警告:
                // Warning: Redis::ping(): connect() failed: Connection refused
                // var_dump('AAAAAAAAA', $redis);
               // echo 'Redis 连接状态' . $redis->ping() . PHP_EOL;
                @$redis->ping();
                if (!$redis->ping()) {
                    goto connect; //跳转到标签出继续执行连接操作
                }
            } catch (Exception  $e) {
                // 信息如 Connection lost 或 Redis server went away
                echo $e->getMessage();
                echo 'Redis 连接失败 重新连接：' . PHP_EOL;
                // 断线重连
                goto connect;
            }
        }
        return $redis;
    }
}

function rlog(...$args)
{
    if (empty($args[0])) {
        return;
    }
    static $LOG_CONSOLE = false; //是否输出到控制台
    static $LOG_NAME = "rfid_station.log"; //值为空时 不写入文件
    static $LOG_SIZE = 64 * 1024 * 1024; //文件最大尺寸

    static $LOG_CACHE = false; //是否缓存日志内容 用于批量写入文件
    static $CACHE_DURATION = 10; //缓存最大时间 秒
    static $CACHE_SIZE = 1024; //缓存大小
    static $cacheStartTime = 0;
    static $cacheBuf = '';

    static $LOG_TIMES = 10; //调用这个函数最大次数 超过次数后判断下文件大小
    static $logCount = 0;


    $buf = '';
    if (count($args) == 1 && $args[0] == "\n") { //只有换行时 不写入时间戳了
        $buf = "\n";
    } else {
        $pid = ''; //进程id
        if (function_exists('posix_getpid')) {
            $pid = ' ' . posix_getpid() . ' ';
        }
        $fileLine = ''; //文件名:行号
        {
            $debug = debug_backtrace();
            $fileLine = ($pid == '' ? ' ' : '') . basename($debug[0]['file']) . ':' . $debug[0]['line'] . ' ';
        }
        $buf = date("y-m-d H:i:s") . "{$pid}{$fileLine}" . implode(' ', $args) . "\n";
    }

    $logCount++;
    if (!empty($LOG_NAME)) {
        if ($LOG_CACHE) {
            $cacheBuf .= $buf;
            //超过缓存尺寸 或者 超过缓存时长 写缓存到文件
            if (strlen($cacheBuf) > $CACHE_SIZE || time() - $cacheStartTime > $CACHE_DURATION) {
                $cacheStartTime = time();
                goto write;
            } else {
                goto skipWrite;
            }
        } else {
            $cacheBuf = $buf;
        }
        write: {
            //超过尺寸后 删除旧文件 把新文件重命名为旧文件  多进程同时操作 不加锁问题不大
            if ($logCount > $LOG_TIMES && filesize($LOG_NAME) > $LOG_SIZE) {
                $oldLogName = $LOG_NAME . '.old';
                if (file_exists($oldLogName)) {
                    if (!unlink($oldLogName)) {
                        echo "unlink err\n";
                    }
                }
                if (!rename($LOG_NAME, $oldLogName)) {
                    echo "rename err\n";
                }
                $logCount = 0;
            }
            if (!file_put_contents($LOG_NAME, $cacheBuf, FILE_APPEND)) {
                echo "file_put_contents err\n";
            }
            $cacheBuf = '';
        }
        skipWrite: {
        }
    }
    if ($LOG_CONSOLE) {
        echo $buf;
    }
}

function http($url, $params, $header = [], $method = 'GET', $timeout = 10)
{
    // POST $params 字符串形式query=abcd&abc=12345
    //GET $params 数组['query' => 'abcd', 'abc' => 12345]
    //  $header[] = "Content-Type: application/x-www-form-urlencoded";
    //  $header[] = "Content-Type: application/soap+xml; charset=utf-8";
    //  $header[] = "Content-Type: application/json; charset=utf-8";
    //  $header[] = "Expect: ";
    rlog("[HTTP] url:$url,method:$method" . ",header:" . json_encode($header));
    if (strtoupper($method) == 'POST') {
        rlog("[POST] send params " . (!is_array($params) ? $params : json_encode($params, JSON_UNESCAPED_UNICODE)));
    } else {
        rlog("[GET] send " . json_encode($params));
    }
    $header[] = "Expect: ";
    $opts = array(
        CURLOPT_TIMEOUT => $timeout,
        CURLOPT_RETURNTRANSFER => 1,
        CURLOPT_SSL_VERIFYPEER => false,
        CURLOPT_SSL_VERIFYHOST => false,
        CURLOPT_HTTPHEADER => $header
    );

    /* 根据请求类型设置特定参数 */
    switch (strtoupper($method)) {
        case 'GET':
            $opts[CURLOPT_URL] = $url . (empty($params) ? '' : ('?' . http_build_query($params)));
            break;
        case 'POST':
            //$params = http_build_query($params);
            $opts[CURLOPT_URL] = $url;
            $opts[CURLOPT_POST] = 1;
            $opts[CURLOPT_POSTFIELDS] = json_encode($params);
            break;
        default:
            rlog("[ERR] method " . $method);
            return false;
    }

    global $ch; //curl长连接
    if (empty($ch)) {
        $ch = curl_init();
    }
    if (empty($ch)) {
        rlog("[ERR] curl_init");
        return false;
    }

    $csa = curl_setopt_array($ch, $opts);
    if (empty($csa)) {
        rlog("[ERR] curl_setopt_array");
        return false;
    }

    $data = curl_exec($ch);
    if ($data === false) {
        rlog("[ERR] curl_exec errno:" . curl_errno($ch) . " " . curl_error($ch));
        return false;
    }
    //unicode转中文
    $data = decodeUnicode($data);
    rlog("[HTTP] recv " . $data);
    //curl_close($ch);
    return $data;
}
function decodeUnicode($str)
{
    return preg_replace_callback('/\\\\u([0-9a-f]{4})/i', function ($matches) {
        return iconv("UCS-2BE", "UTF-8", pack("H*", $matches[1]));
    }, $str);
}
function loop()
{
    // $server   = '43.157.183.167';
    // $port     = 1883;
    // $clientId = 'local1_mqtt_livestock_cli_202406271';
    // $username = 'rl517';
    // $password = "rlian2022";

    $server   = '47.114.189.154';
    $port     = 1883;
    $clientId = 'local_mqtt_livestock_cli_202501271';
    $username = 'rl241107';
    $password = "rlian2024";

    $clean_session = false;

    $connectionSettings  = new ConnectionSettings();
    $connectionSettings = $connectionSettings
        ->setUsername($username)
        ->setPassword($password)
        ->setKeepAliveInterval(60)
        // Last Will 设置
        //        ->setLastWillTopic('emqx/test/last-will')
        //        ->setLastWillMessage('client disconnect')
        //        ->setLastWillQualityOfService(1)
    ;

    //include "RLog.php";
    //    $mqtt = new MqttClient($server, $port, $clientId, MqttClient::MQTT_3_1, null, new RLog());

    $mqtt = new MqttClient($server, $port, $clientId);

    $mqtt->connect($connectionSettings, $clean_session);
    rlog('INFO', "connect OK");

    /*
    消息方向	设备->服务器 
    设备主动上报当前设备公共信息参数：ScBusTem/DevRegularInfo
    服务器获取设备系统信息后设备上传信息，即GetDevSysMsg的回应 ScBusTem/GetUpDevSysMsg 
    服务器设置设备重量信息信息 ScBusTem/RCInfoMsg
    */

    //终端上报系统信息数据
    $mqtt->subscribe('RL/DATA/#', function ($topic, $message) use($mqtt) {
        // rlog("reportData", 'recv', $topic, $message);
        // $topicArr=explode('/',$topic);
        $data=bin2hex($message);
        rlog("reportData", 'recv', $topic, $data);
        //9cbc01080386002098c42af3000003011865a741200102e798c42c01110003012165a7412001fc0e
        // $str="9cbc01080386002098c42af3000003011865a741200102e798c42c01110003012165a7412001fc0e";
        $parsedData = parseData($data);
        rlog("reportData", 'pares', json_encode($parsedData));
        mqttToRedis(json_encode($parsedData));



    }, 1);

    $mqtt->loop(true);
}
function parseData($dataStr) {
    // 将十六进制字符串转换为二进制
    //9cbc010803860020 98c42af3000003011865a741200102e7 98c42c01110003012165a7412001fc0e
    // 定义解析结果数组
    $result = [];
    // 解析固定部分
    $result['magic'] = substr($dataStr, 0, 4);  // 9cbc 固定
    $result['type'] =substr($dataStr, 4, 2);  // 01 固定
    // 解析 MAC 地址
    $result['mac'] = substr($dataStr, 6, 6);  // 080386
    // 解析长度
    $result['length'] = hexdec(substr($dataStr, 12, 4)) *2;  // 0020 (32 bytes)
    $data=[];
    $data_str=substr($dataStr, 16);
    if(strlen($data_str)!=$result['length']){
        echo '数据格式不正确';
    }
    for($i=0;$i<$result['length'];$i+=32){
        $unit = substr($data_str,$i,32);
        $item=[];
        // 标签号
        $item['seq'] =(string)hexdec(substr($unit, 0, 6));  // 98c42a
        // 步数逆序（2字节）
        $binaryData = hex2bin(substr($unit, 6, 6));// f30000 -> 步数 243
        // 将字节数据逆序排列
        $reversedBinaryData = strrev($binaryData);
        // 将逆序后的二进制数据转换为十六进制
        $reversedHex = bin2hex($reversedBinaryData);
        // 步数
        $item['steps'] = hexdec($reversedHex);
        // 保留字段
        $item['reserve'] = substr($unit, 12, 2);  // 03
        // 状态
        $item['stat'] = substr($unit, 14, 2);  // 01 (告警状态-正常)
        // 信号强度
        $item['rssi'] = hexdec(substr($unit, 16, 2));  // 18
        // 时间戳（4字节）
        $item['timestamp'] = hexdec(substr($unit, 18, 8));  // 65a74120 (Unix时间戳)
        $item['datetime'] =date('Y-m-d H:i:s',$item['timestamp']); 
        // 收到次数
        $item['rec_count'] = substr($unit, 26, 2);  // 01
        // 温度（2字节，除以100）
        $hexPositive = substr($unit, 28, 4);  // 02e7 -> 7.43度
        $positiveDecimal = hexdec($hexPositive); // 743
        if ($positiveDecimal >= 0x8000) {  // 如果数值大于等于 0x8000, 表示负数
            $positiveDecimal -= 0x10000;  // 进行补码转换
        }
        $item['temp'] =round($positiveDecimal / 100,2); // 7.43
        $data[]=$item;
    }
    $result['data']=$data;


    return $result;
}
function mqttToRedis($text){
    
    try{
        
        
         app_redis()->lpush("station_rfid_eartag_data",$text);

    }catch(Exception $e){
        rlog("INFO", 'recv',"redis 异常".$e->getMessage());
    }
    
} 


while (1) {
    try {
        rlog('INFO', 'connect start');
        loop();
    } catch (\Exception $ex) {
        rlog("ERR", $ex->getTraceAsString());
        rlog("ERR", $ex->getMessage());
    }
    sleep(3);
}


// $str="9cbc010700300180004ea639000013015a67778bdf04024d004ea25a000013015667778bdc2303c0004e943a000013015a67778b52020cb3004e9f3e000013015967778b98080c30004e9c1c000013015767778a890403ca004e8f4f000013015967778bd90503d4004ea323010013015a67778b84030463004e9b62000013015c67778b8c08033f004e995c000013015867778bd414048c004e9382000013015a67778bdf0704c0004e985b000013015867778bde31058b004ea035000013015a67778b01040105004e924d000013015467778be13a0187004ea509000013015a67778b24010722004e9736000013015267778a710f01a3004e9634000013015a67778b91140716004e91180000130155677789db010ab2004e9d35000013015d67778baa0e06d0004ea93f000013015e67778bd302043a004e9055000013015a67778bd70f017e004eaae3000013015867778b3b0201c8004e9e63000013015c67778b0d0203f2004e9a2a000013014d67778be31f02b6004ea122000013015867778a420608c5";
// $parsedData = parseData($str);
// var_dump($parsedData);