當前位置：首頁 >

PM2.5检测 -- PMS7003 采集和 MQTT 传输

發(fā)布時間：2025/3/15 34 豆豆

生活随笔收集整理的這篇文章主要介紹了 PM2.5检测 -- PMS7003 采集和 MQTT 传输小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

下一篇文章里我們通過交叉編譯生成了?libmosquitto.so.1?

參看：MQTT再學習 -- 交叉編譯與移植

之前有講過MQTT客戶端的測試，參看：MQTT再學習 -- 安裝MQTT客戶端及測試

那接下來，我們就開始先測試一下，看看交叉編譯生成的客戶端，是否能發(fā)送數(shù)據(jù)到服務(wù)器上。

一、生成交叉編譯客戶端

先看一下，需要的源碼文件，這個在客戶端測試那篇文章里有提供源碼的哦。 # ls a.txt deviceCfg.xml libmosquitto.so.1 mosq.c mosquitto.h net_zslf.c net_zslf.h readme.txt 交叉編譯，生成目標文件。
arm-none-linux-gnueabi-gcc -c net_zslf.c mosq.c?
鏈接共享庫： arm-none-linux-gnueabi-gcc mosq.o net_zslf.o -L . libmosquitto.so.1 /opt/com/ssl/lib/libssl.so.1.0.0 /opt/com/ssl/lib/libcrypto.so.1.0.0 -o client
生成可執(zhí)行文件 client //查看文件屬性 # file client client: ELF 32-bit LSB executable, ARM, version 1 (SYSV), dynamically linked (uses shared libs), for GNU/Linux 2.6.14, not stripped//Ubuntu下不能執(zhí)行 # ./client bash: ./client: cannot execute binary file

注意： ?這個必須要鏈接 libssl.so.1.0.0 libcrypto.so.1.0.0，否則出現(xiàn)錯誤：

/opt/arm-2009q1-203/bin/../lib/gcc/arm-none-linux-gnueabi/4.3.3/../../../../arm-none-linux-gnueabi/bin/ld: cannot find -lmosquitto
collect2: ld returned 1 exit status

這部分參看文章：

mosquitto的交叉編譯mips

mqtt的開源實現(xiàn)mosquitto的交叉編譯與使用?

嘗試了一上午終于找到原因不容易啊。至此生成了Linux下可執(zhí)行的客戶端的二進制文件 client

二、在開發(fā)板上執(zhí)行客戶端測試

將這三個庫文件 libcrypto.so.1.0.0 libmosquitto.so.1 libssl.so.1.0.0 ?拷貝到開發(fā)板 lib 目錄下

然后拷貝 client 和?deviceCfg.xml 放在同一目錄

執(zhí)行client 之前需要修改權(quán)限：chmod 777 client?

Ubuntu下打開 MQTT 服務(wù)器

開發(fā)板下執(zhí)行 client

OK，終于完成了！！！

證明交叉編譯生成的客戶端測試代碼是可行的。

三、客戶單測試源碼下載

下載：交叉編譯 MQTT 客戶端測試源碼

四、結(jié)合MQTT和PMS7003開發(fā)程序

這才是本篇文章的核心，也是本項目最后的一步了。

參看：PM2.5檢測 -- PMS7003 開發(fā)程序 #include <fcntl.h> //文件控制定義 #include <stdio.h> //標準輸入輸出定義 #include <stdlib.h> //標準函數(shù)庫定義 #include <unistd.h> //Unix標準函數(shù)定義 #include <errno.h> //錯誤好定義 #include <termios.h> //POSIX終端控制定義 #include <sys/ioctl.h> //ioctl函數(shù)定義 #include <string.h> //字符操作 #include <sys/types.h> #include <sys/stat.h> #include <pthread.h> #include <sys/time.h> #include <sys/msg.h> #include <stdint.h> #include <stdbool.h> #include <pthread.h> #include "mosquitto.h" #include "net_zslf.h" #include <sys/wait.h> //message queue variable int message_count = 0; //const char *mqtt_broker_address = "192.168.2.52"; /* mqtt_broker ip address */ int mqtt_broker_port = 1883; /* mqtt_broker port number */ long msgtype = 10; /* pm sensor message type */ //int msgsize = 100; /* pm sensor message size */ int msgsize = sizeof (MSG_data_buf); /* pm sensor message size */ int fd_gpio; struct termios newtio, oldtio; typedef struct { int pin_idx; int pin_dir; int pin_sta; } davinci_gio_arg; typedef enum { AT91PIO_DIR_OUT = 0, AT91PIO_DIR_INP } davinci_gio_dir; //驅(qū)動判斷輸入輸出模式 davinci_gio_arg arg; #define DEV_PIO_LED "/dev/pio" // 需要手動添加設(shè)備號 mknod /dev/pio c 203 0 #define PIO_NUM 47 // 47pin 為控制輸入輸出方向引腳 #define DEV_UART "/dev/ttyS1" // /dev/ttyS1 為串口設(shè)備 #define IOCTL_PIO_SETDIR 1 //set gpio direct #define IOCTL_PIO_GETDIR 2 //get gpio direct #define IOCTL_PIO_SETSTA 3 //set gpio status #define IOCTL_PIO_GETSTA 4 //get gpio status //保存信息 int log_init( const char *strFileName ) { int fdLog = -1; if( -1 == (fdLog = open( strFileName, O_CREAT|O_TRUNC ) ) ) { } close( fdLog ); } int log_out( const char *strFileName, const char * szLog ) { int fdLog = -1; if( -1 == ( fdLog = open( strFileName, O_CREAT|O_WRONLY|O_APPEND ) ) ) { printf( "LOG (%s) open error!\n", strFileName ); return -1; } write( fdLog, szLog, strlen( szLog ) ); close( fdLog ); return 0; } //配置串口 /* 參數(shù)說明：fd 設(shè)備文件描述符，nspeed 波特率，nbits 數(shù)據(jù)位數(shù)（7位或8位）， parity 奇偶校驗位（'n'或'N'為無校驗位，'o'或'O'為偶校驗，'e'或'E'奇校驗）， nstop 停止位（1位或2位）成功返回1，失敗返回-1。 */ int set_com_opt( int fd, int nspeed, int nbits, char parity, int nstop ) { char szTmp[128]; //打印配置信息 sprintf( szTmp, "set_com_opt - speed:%d,bits:%d,parity:%c,stop:%d\n", nspeed, nbits, parity, nstop ); log_out( "./485.log", szTmp ); //保存并測試現(xiàn)在有串口參數(shù)設(shè)置，在這里如果串口號等出錯，會有相關(guān)的出錯信息 if( tcgetattr( fd, &oldtio ) != 0 ) { sprintf( szTmp, "SetupSerial 1" ); log_out( "./485.log", szTmp ); perror( "SetupSerial 1" ); return -1; } //修改輸出模式，原始數(shù)據(jù)輸出 bzero( &newtio, sizeof( newtio )); newtio.c_cflag &=~(OPOST); //屏蔽其他標志位 newtio.c_cflag |= (CLOCAL | CREAD ); newtio.c_cflag &= ~CSIZE; //設(shè)置數(shù)據(jù)位 switch( nbits ) { case 7: newtio.c_cflag |= CS7; break; case 8: newtio.c_cflag |= CS8; break; default: perror("Unsupported date bit!\n"); return -1; } //設(shè)置校驗位 switch( parity ) { case 'n': case 'N': //無奇偶校驗位 newtio.c_cflag &= ~PARENB; newtio.c_iflag &= ~INPCK; break; case 'o': case 'O': //設(shè)置為奇校驗 newtio.c_cflag |= ( PARODD | PARENB ); newtio.c_iflag |= ( INPCK | ISTRIP ); break; case 'e': case 'E': //設(shè)置為偶校驗 newtio.c_iflag |= ( INPCK |ISTRIP ); newtio.c_cflag |= PARENB; newtio.c_cflag &= ~PARODD; break; default: perror("unsupported parity\n"); return -1; } //設(shè)置停止位 switch( nstop ) { case 1: newtio.c_cflag &= ~CSTOPB; break; case 2: newtio.c_cflag |= CSTOPB; break; default : perror("Unsupported stop bit\n"); return -1; } //設(shè)置波特率 switch( nspeed ) { case 2400: cfsetispeed( &newtio, B2400 ); cfsetospeed( &newtio, B2400 ); break; case 4800: cfsetispeed( &newtio, B4800 ); cfsetospeed( &newtio, B4800 ); break; case 9600: cfsetispeed( &newtio, B9600 ); cfsetospeed( &newtio, B9600 ); break; case 115200: cfsetispeed( &newtio, B115200 ); cfsetospeed( &newtio, B115200 ); break; case 460800: cfsetispeed( &newtio, B460800 ); cfsetospeed( &newtio, B460800 ); break; default: cfsetispeed( &newtio, B9600 ); cfsetospeed( &newtio, B9600 ); break; } //設(shè)置等待時間和最小接收字符 newtio.c_cc[VTIME] = 0; newtio.c_cc[VMIN] = 0; //VTIME=0，VMIN=0，不管能否讀取到數(shù)據(jù)，read都會立即返回。 //輸入模式 newtio.c_lflag &= ~(ICANON|ECHO|ECHOE|ISIG); //設(shè)置數(shù)據(jù)流控制 newtio.c_iflag &= ~(IXON|IXOFF|IXANY); //使用軟件流控制 //如果發(fā)生數(shù)據(jù)溢出，接收數(shù)據(jù)，但是不再讀取刷新收到的數(shù)據(jù)但是不讀 tcflush( fd, TCIFLUSH ); //激活配置 (將修改后的termios數(shù)據(jù)設(shè)置到串口中） if( tcsetattr( fd, TCSANOW, &newtio ) != 0 ) { sprintf( szTmp, "serial set error!\n" ); log_out( "./485.log", szTmp ); perror( "serial set error!" ); return -1; } log_out( "./485.log", "serial set ok!\n" ); return 1; } //打開串口并返回串口設(shè)備文件描述 int open_com_dev( char *dev_name ) { int fd; char szTmp[128]; log_init( "./485.log" ); if(( fd = open( dev_name, O_RDWR|O_NOCTTY|O_NDELAY)) == -1 ) { perror("open\n"); //printf("Can't open Serial %s Port!\n", dev_name ); sprintf( szTmp, "Can't open Serial %s Port!\n", dev_name ); log_out( "./485.log", szTmp ); return -1; } sprintf( szTmp, "open %s ok!\n", dev_name ); log_out( "./485.log", szTmp ); if(fcntl(fd,F_SETFL,0)<0) { printf("fcntl failed!\n"); } //printf("Open %s ok\n",dev_name ); return fd; } int main (void) { int gflags; int msgid; key_t key; pthread_t thread1, thread2; int ret; /* struct msqid_ds msg_ginfo, msg_sinfo; */ char *msgpath = "home/tarena/project/MQTT/test/a.txt"; //消息隊列的鍵 key = ftok(msgpath, 'a'); gflags = IPC_CREAT; //創(chuàng)建消息隊列 msgid = msgget(key, gflags | 00666); if(msgid == -1) { DUG_PRINTF("msg create error\n"); return -1; } int error = 0, error1 = 0; arg.pin_idx = PIO_NUM; arg.pin_dir = AT91PIO_DIR_OUT; //打開/dev/pio設(shè)備 fd_gpio = open(DEV_PIO_LED, O_RDWR); if(fd_gpio < 0) { perror("fd_gpio open err"); exit (-1); } /* msg_stat(msgid,msg_ginfo); */ //創(chuàng)建消息隊列發(fā)送線程 ret = pthread_create(&thread1, NULL, &start_thread_msgsend, (void *)&msgid); if (ret != 0) { perror("pthread msgsend create error\n"); return -1; } //創(chuàng)建消息隊列接收線程 ret = pthread_create(&thread2, NULL, &start_thread_msgrcv, (void *)&msgid); if (ret != 0){ perror("pthread msgrcv create error\n"); return -1; } //線程等待 pthread_join(thread1, NULL); pthread_join(thread2, NULL); //關(guān)閉設(shè)備 close (fd_gpio); return 0; } //消息隊列接收線程 void *start_thread_msgrcv(void *arg) { int rflags = 0; int ret; int msgid = *(int *)(arg); MSG_data_buf msg_rbuf; //消息隊列類型 struct mosquitto *mosq; //保存一個MQTT客戶端連接的所有信息 //下面的代碼是從xml文件中讀取 FILE *fp; char szFileBuff[1024] = {0}; char serverADDR[16] = {0},devID[15] = {0},devName[15] = {0}; char Longitude[15] = {0},Latitude[15] = {0},frequency[3] = {0}; char *lFirst, *lEnd; char devInfo[70]; FILE *fp_re; char buffer_re[4]; //打開xml文件 fp = fopen("deviceCfg.xml","r"); if (fp==NULL) { DUG_PRINTF("read XML file error!\n"); } //你只要知道while里面是獲取xml信息的，至于這種操作有點6 while(fgets(szFileBuff, 1023, fp)) { if ((lFirst = strstr(szFileBuff, "<serverADDR>")) != NULL) { lEnd = strstr(lFirst + 1, "</serverADDR>"); memcpy(serverADDR, lFirst + 12, lEnd - lFirst - 12); } if ((lFirst = strstr(szFileBuff, "<devID>")) != NULL) { lEnd = strstr(lFirst + 1, "</devID>"); memcpy(devID, lFirst + 7, lEnd - lFirst - 7); } if ((lFirst = strstr(szFileBuff, "<devName>")) != NULL) { lEnd = strstr(lFirst + 1, "</devName>"); memcpy(devName, lFirst + 9, lEnd - lFirst - 9); } if ((lFirst = strstr(szFileBuff, "<Longitude>")) != NULL) { lEnd = strstr(lFirst + 1, "</Longitude>"); memcpy(Longitude, lFirst + 11, lEnd - lFirst - 11); } if ((lFirst = strstr(szFileBuff, "<Latitude>")) != NULL) { lEnd = strstr(lFirst + 1, "</Latitude>"); memcpy(Latitude, lFirst + 10, lEnd - lFirst - 10); } //下面這個語句是用于分頻率傳送數(shù)據(jù)的 if ((lFirst = strstr(szFileBuff, "<frequency>")) != NULL) { lEnd = strstr(lFirst + 1, "</frequency>"); memcpy(frequency, lFirst + 11, lEnd - lFirst - 11); } if ((lFirst = strstr(szFileBuff, "</display>")) != NULL) { sprintf(devInfo, "&%s&%s&%s&%s&\n",devID,devName,Longitude,Latitude); } } fclose(fp); // sleep(18); //這里是關(guān)鍵了，MQTT協(xié)議要上演了這部分為 pub 發(fā)布內(nèi)容 //MQTT 庫初始化 mosquitto_lib_init(); //新建 mosq = mosquitto_new(devID, true, NULL); //連接回調(diào)設(shè)置 mosquitto_connect_callback_set(mosq, my_connect_callback); //斷開回調(diào)設(shè)置 mosquitto_disconnect_callback_set(mosq, my_disconnect_callback); //發(fā)布回調(diào)設(shè)置 mosquitto_publish_callback_set(mosq, my_publish_callback); //MQTT連接 if(mosquitto_connect(mosq, serverADDR, mqtt_broker_port, 600) != MOSQ_ERR_SUCCESS) { DUG_PRINTF("mosquitto connection error\n"); //銷毀 mosquitto_destroy(mosq); //清空 mosquitto_lib_cleanup(); } // sleep(18); //這里開始消息隊列接收消息 while (1) { //接收消息 ret=msgrcv(msgid, &msg_rbuf, msgsize, msgtype, rflags); //sleep(1); if (ret == -1) { DUG_PRINTF("read msg error\n"); } DUG_PRINTF("%s\n", msg_rbuf.mtext); <span style="white-space:pre"> </span>//strcat(msg_rbuf.mtext,devInfo); 將其和其他信息合并 //將接收到的信息發(fā)布 mosquitto_publish(mosq, NULL, "pmsensor", ret, (void *)msg_rbuf.mtext, 0, 0); //sleep(5); <span style="white-space:pre"> </span>memset (msg_rbuf.mtext, 0, sizeof (msg_rbuf.mtext)); } /* should never run below */ //銷毀 mosquitto_destroy(mosq); //清空 mosquitto_lib_cleanup(); return NULL; } //單片機數(shù)據(jù)收發(fā) void* start_thread_msgsend(void* p) { char buf[64]; char frameBuf[64]; int detectOff = 0; int res = 0, nread = 0; int m_pm1_factory; int m_pm25_factory; int m_pm10_factory; int m_pm1_outdoor; int m_pm25_outdoor; int m_pm10_outdoor; int m_count03; int m_count05; int m_count1; int m_count25; int m_count5; int m_count10; unsigned short m_length; unsigned short m_version; unsigned short m_errorno; char pm_f[30]; int retval; MSG_data_buf msg_sbuf; //消息隊列類型 int msgid = *(int *)(p); int sflags=IPC_NOWAIT; msg_sbuf.mtype = msgtype; while (1) { arg.pin_sta = 0; //設(shè)為低電平接收態(tài) ioctl(fd_gpio, IOCTL_PIO_SETSTA, &arg); int fd_r=open_com_dev( DEV_UART ); if( fd_r < 0 ) { printf( "open UART device error! %s\n", DEV_UART ); } else set_com_opt(fd_r, 9600,8,'n',1); //執(zhí)行select fd_set rd; FD_ZERO(&rd); FD_SET(fd_r, &rd); if ((res = select (fd_r+1,&rd, NULL, NULL, NULL) )< 0) { perror ("read err"); exit (-1); } memset (buf, 0, sizeof (buf)); if (FD_ISSET (fd_r, &rd)) { //接收數(shù)據(jù) 8 8 2 int res1 = 0; int val = 0; int calcChecksum = 0; int checksum = 0; int i = 0; while ((nread = read(fd_r, buf, 1)) > 0) { //printf ("%02X ",*buf); //frameBuf[detectOff] = buf[0]; memcpy (frameBuf+detectOff, buf, 1); //calcChecksum += *buf; detectOff++; if (frameBuf[0] == 0x42 && frameBuf[1] == 0x4d) { // m_length = frameBuf[3]+(frameBuf[2]<<8); m_pm1_factory = frameBuf[5]+(frameBuf[4]<<8); m_pm25_factory = frameBuf[7]+(frameBuf[6]<<8); m_pm10_factory = frameBuf[9]+(frameBuf[8]<<8); m_pm1_outdoor = frameBuf[11]+(frameBuf[10]<<8); m_pm25_outdoor = frameBuf[13]+(frameBuf[12]<<8); m_pm10_outdoor = frameBuf[15]+(frameBuf[14]<<8); // m_count03 = frameBuf[17]+(frameBuf[16]<<8); // m_count05 = frameBuf[19]+(frameBuf[18]<<8); // m_count1 = frameBuf[21]+(frameBuf[20]<<8); // m_count25 = frameBuf[23]+(frameBuf[22]<<8); // m_count5 = frameBuf[25]+(frameBuf[24]<<8); // m_count10 = frameBuf[27]+(frameBuf[26]<<8); // m_version = frameBuf[28]; // m_errorno = frameBuf[29]; // checksum = frameBuf[31]+(frameBuf[30]<<8); // calcChecksum -= ((checksum>>8)+(checksum&0xFF)); } //退出循環(huán)，這里有點疑問 if (detectOff == 32) { //printf ("\n"); //printf ("pm1_factory = %d ug/m3\npm25_factory = %d ug/m3\npm10_factory = %d ug/m3\n", m_pm1_factory, m_pm25_factory, m_pm10_factory); //printf ("pm1_outdoor = %d ug/m3\npm25_outdoor = %d ug/m3\npm10_outdoor = %d ug/m3\n",m_pm1_outdoor,m_pm25_outdoor,m_pm10_outdoor); // printf ("m_count03 = %02X\nm_count05 = %02X\nm_count1 = %02X\nm_count25 = %02X\nm_count5 = %02X\nm_count10 = %02X\n", // m_count03, m_count05, m_count1, m_count25, m_count5, m_count10); // printf ("m_length = %d\n", m_length); // printf ("m_version = %s\n", m_version); // printf ("m_errorno = %s\n", m_errorno); // printf("checksum = %02X %s calcChecksum = %02X",checksum, (calcChecksum == checksum ? "==" : "!="), calcChecksum); // for (i = 0;i<32;i++) // { // printf ("%02X ", frameBuf[i]); // } // printf ("\n"); sprintf(pm_f,"%d %d",m_pm25_factory,m_pm10_factory); /*這里和下面的 read 結(jié)合就好了 */ strcpy (msg_sbuf.mtext,pm_f); //strcpy (msg_sbuf.mtext,"hello world"); //消息隊列發(fā)送 //retval = msgsnd(msgid, &msg_sbuf,msgsize, sflags); retval = msgsnd(msgid, &msg_sbuf,msgsize, 0); if(retval == -1) { DUG_PRINTF("message send error\n"); } memset (frameBuf, 0, sizeof (frameBuf)); memset (pm_f, 0, sizeof (pm_f)); memset (msg_sbuf.mtext, 0, sizeof (msg_sbuf.mtext)); detectOff = 0; break; } } } close (fd_r); usleep (200000); } } 測試結(jié)果： Ubuntu下打開 MQTT 服務(wù)器

開發(fā)板下執(zhí)行 client