基于阿里云MQTT物联网平台视频监控(下)
1.项目介绍
本项目基于物联量平台远程的视频监控项目，通过mqtt协议实现两个设备间的数据上报与订阅。通过这个项目来演示，两个mqtt设备如何互相订阅，进行消息流转。在阿里云服务器上创建2个设备，分为为设备a和设备b；设备a负责采集本地摄像头画面上传，设备b负责接收设备a上传的数据然后解析显示出来。在阿里云服务器上需要配置云产品流转，让设备a的数据上传后自动发送给设备b。这样就完成了视频画面数据的流转。不过因为阿里云的最大数据限制，每次最大发送10240字节的数据。
2.linux 下 socket 编程连接阿里云物联网平台
#define server_ip asfdda.iot-as-mqtt.cn-shanghai.aliyuncs.com//服务器ip#define server_port 1883 //端口号#define clientid aasfsaxabf.imasfas|securemode=2,signmethod=hmacsha256,timestamp=1678323607797|#define username imsfea&a1sadf8xabf#define password 15566ab496e81da728a3792ebe532fd4a3f4026a2b831df5af24da06//密文 #define set_topic /sys/a14dxabf/imagfa/thing/service/property/set //订阅 #define post_topic /sys/a14sdf8xabf/imdfea/thing/event/property/post //发布 int main(){ pthread_t id; signal(sigpipe,sig_ign);/*忽略sigpipe信号*/ signal(sigalrm,signal_func);/*闹钟信号*/ sockfd=socket(af_inet,sock_stream,0); if(sockfd==-1) { printf(网络套接字打开失败n); return 0; } /*设置发送缓冲区大小*/ int nsendbuf=40*1024;//设置为 20k if(setsockopt(sockfd,sol_socket,so_sndbuf,(const char*)&nsendbuf,sizeof(int))) { printf(setsockopt(so_sndbuf) 设置错误!n); return 0; } /*域名解析*/ struct hostent *hostent; while(1) { hostent=gethostbyname(server_ip); if(hostent==null) { printf(域名解析失败n); sleep(1); } else break; } printf(主机名:%sn,hostent->h_name); printf(协议类型:%sn,(hostent->h_addrtype == af_inet)?af_inet:af_inet6); printf(ip地址长度:%dn,hostent->h_length); char *ip; for(int i=0;hostent->h_addr_list[i];i++) { ip=inet_ntoa(*(struct in_addr *)hostent->h_addr_list[i]); printf(ip=%sn,ip); } /*连接服务器*/ struct sockaddr_in addr; addr.sin_family=af_inet;//ipv4 addr.sin_port=htons(server_port);/*端口号*/ addr.sin_addr.s_addr=inet_addr(ip);//服务器ip if(connect(sockfd, (struct sockaddr *)&addr,sizeof(struct sockaddr_in))==0) { printf(服务器连接成功n); while(1) { mqtt_init(); /*登录服务器*/ if(mqtt_connect(clientid,username,password)==0) { break; } sleep(1); printf(服务器连接中....n); } printf(连接成功rn); //订阅物联网平台数据 stat=mqtt_subscribetopic(set_topic,1,1); if(stat) { close(sockfd); printf(订阅失败rn); exit(0); } printf(订阅成功rn); /*创建线程*/ pthread_create(&id, null,pth_work_func,null); pthread_detach(id);//设置分离属性 alarm(3);//闹钟函数，时间到达会产生sigalrm信号 int a=0; while(1) { sprintf(mqtt_message,{method:thing.event.property.post,params:{image:阿里云物联网平台测试}}); mqtt_publishdata(post_topic,mqtt_message,0);//发布数据 } }}
3.云产品流转
云产品流转文档： https: //help. aliyun. com/document_detail/68677. html
3.1 什么是云产品流转
设备基于 topic 与物联网平台进行通信时， 您可以在数据流转中， 编写 sql 对 topic 中的数据进行处理， 并配置转发规则将处理后的数据转发到其他设备 topic 或阿里云其他服务。
3.2 云产品流转配置
1 .创建解析器
2.关联数据源
3.关联数据目的
4.编写解析器脚本
解析器说明文档： https: //help. aliyun. com/document_detail/270931. html
格式示例：
//通过 payload 函数， 获取设备上报的消息内容， 并按照 json 格式转换。var data = payload(json); //直接流转物模型上报数据。writeiottopic(1000, topic, data);
topic 如下：
编辑好后发布即可， 至此， 阿里物联网平台配置完成。
4.代码实现
4.1 设备 a 发送方
1 .usb 摄像头应用编程
采用 linux 下 v4l2 框架初始化 usb 摄像头， 采集图像数据。
/*摄像头初始化返回值：成功返回摄像头描述符，失败返回负数*/int video_init(struct camera *camera){ int video_fd; int i=0; /*1.打开设备节点*/ video_fd=open(video_dev,o_rdwr); if(video_fd==-1)return -1; /*2.设置摄像头格式*/ struct v4l2_format format; memset(&format,0,sizeof(format)); format.type=v4l2_buf_type_video_capture;//视频捕获格式 format.fmt.pix.width=320; format.fmt.pix.height=240; format.fmt.pix.pixelformat=v4l2_pix_fmt_yuyv;//图像数据格式yuyv if(ioctl(video_fd,vidioc_s_fmt,&format))return -2; printf(图像尺寸:%d * %dn,format.fmt.pix.width,format.fmt.pix.height); camera->image_w=format.fmt.pix.width; camera->image_h=format.fmt.pix.height; /*3.向内核请求缓冲区*/ struct v4l2_requestbuffers reqbuf; memset(&reqbuf,0,sizeof(reqbuf)); reqbuf.count=4;/*缓冲区个数*/ reqbuf.type=v4l2_buf_type_video_capture;//视频捕获格式 reqbuf.memory=v4l2_memory_mmap;/*内存映射*/ if(ioctl(video_fd,vidioc_reqbufs,&reqbuf))return -3; printf(缓冲区个数:%dn,reqbuf.count); /*4.将缓冲区映射到进程空间*/ struct v4l2_buffer quebuff; for(i=0;imamp_buff[i]=mmap(null,quebuff.length,prot_read|prot_write,map_shared,video_fd,quebuff.m.offset); printf(buff[%d]=%pn,i,camera->mamp_buff[i]); camera->mmap_size=quebuff.length; } /*5.将缓冲区添加到采集队列*/ for(i=0;i
2.图片编码处理
实时采集图像数据， 将图片数据编码为 jpg 图像格式， 再进 base64 格式编码。base64 编码是一种将二进制数据转换为 ascii 字符的方法， 它使用 64 个字符来表示任意序列的二进制数据。 base64 编码后的数据长度会比原始二进制数据略长， 但可以方便地被转换为文本格式并在网络上进行传输。
3.base64 格式编码
#include static const char * base64char = abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz0123456789+/;/*函数功能：将图片编码为base64格式形参：bindata 源图片数据 base64 编码后的数据 binlength --源文件大小返回值：返回编码后的base64数据*/char * base64_encode( const unsigned char * bindata, char * base64, int binlength ){ int i, j; unsigned char current; for ( i = 0, j = 0 ; i > 2) ; current &= (unsigned char)0x3f; base64[j++] = base64char[(int)current]; current = ( (unsigned char)(bindata[i] > 4) ) & ( (unsigned char) 0x0f ); base64[j++] = base64char[(int)current]; current = ( (unsigned char)(bindata[i+1] > 6) ) & ( (unsigned char) 0x03 ); base64[j++] = base64char[(int)current]; current = ( (unsigned char)bindata[i+2] ) & ( (unsigned char)0x3f ) ; base64[j++] = base64char[(int)current]; } base64[j] = ''; return base64;}
4. base64 格式解码
/*函数功能：base64格式数据解码形参：base64 base64格式数据 bindata 保存解码成功的图像数据返回值：成功返回解码的图像大小*/static const char * base64char = abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz0123456789+/;int base64_decode( const char * base64, unsigned char * bindata ){ int i, j; unsigned char k; unsigned char temp[4]; for ( i = 0, j = 0; base64[i] != '' ; i += 4 ) { memset( temp, 0xff, sizeof(temp) ); for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i] ) temp[0]= k; } for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i+1] ) temp[1]= k; } for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i+2] ) temp[2]= k; } for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i+3] ) temp[3]= k; } bindata[j++] = ((unsigned char)(((unsigned char)(temp[0] 4)&0x03)); if ( base64[i+2] == '=' ) break; bindata[j++] = ((unsigned char)(((unsigned char)(temp[1] 2)&0x0f)); if ( base64[i+3] == '=' ) break; bindata[j++] = ((unsigned char)(((unsigned char)(temp[2]width=cinfo.output_width; unsigned char *rgb_data=image_rgb->rgb; /*为一条扫描线上的像素点分配存储空间，一行一行的解码*/ int row_stride = cinfo.output_width * cinfo.output_components; buffer = (unsigned char *)malloc(row_stride); //将图片内容显示到framebuffer上,cinfo.output_scanline表示当前行的位置，读取数据是会自动增加 i=0; while(cinfo.output_scanline width); //printf(源图片高:%dn,image_rgb->height); u32 w=image_rgb->width; u32 h=image_rgb->height; u8 *src_rgb=image_rgb->rgb;//源图片rgb值 unsigned long oneline_byte=w*3;//一行字节数 float zoom_count=0; /*按比例缩放*/ zoom_count=(lcd_width/(w*1.0)) > (lcd_hight/(h*1.0)) ? (lcd_hight/(h*1.0)):(lcd_width/(w*1.0)); int new_w,new_h; new_w=zoom_count*w;//新图片宽 new_h=zoom_count*h;//新图片高 //printf(新图片宽:%dn,new_w); //printf(新图片高:%dn, new_h); //printf(缩放比例:%.0f%%n,(new_w*1.0/w)*100); unsigned long new_oneline_byte=new_w*3; unsigned char *newbmp_buff=(unsigned char *)malloc(new_h*new_oneline_byte);//动态分配新图片rgb颜色数据缓冲区 if(newbmp_buff==null) { printf([%s line %d]动态分配空间失败n,__function__,__line__); return -1; } memset(newbmp_buff, 0, new_h*new_oneline_byte); /************************图像处理算法(双线性插值)*******************************/ int i,j; for(i=0;irgb,newbmp_buff,new_h*new_oneline_byte);//新图像rgb数据 image_rgb->width=new_w;//新图像宽 image_rgb->height=new_h;//新图像高 free(newbmp_buff); return 0;}
4.3 项目效果
电力线载波报警器,power line alarm
王阳元院士：新变化的历史时期，如何抓住机遇、应对挑战？
八路全双工电力线载波对讲机,Power line interphone
暖通空调设备远程智能监控解决方案
乐视最新消息：降价促销+40亿售世茂广场+4.2亿地皮撂荒+乐视员工求职难
基于阿里云MQTT物联网平台视频监控(下)
介绍两款高性能DC-40GHz的一分二电阻式功分器
追觅、小狗、戴森吸尘器对比测评，让你了解哪款最好用
Lumia 950 XL、Surface Go、苹果 M1 MacBook 跑上 Win10X ：后者性能尤其好
stm32f407浮点运算速度
Vishay推出背面绝缘的TrenchFET功率MOSFET
CyclicBarrier 任务实践
荣耀V30 PRO的多摄协同录制功能怎么样
智能物联网企业触景无限入选2023年度高成长企业TOP100榜单
XG1590射频、中频、音频放大电路的应用
自动驾驶商业落地现状与未来发展前景
中国移动与诺基亚达成10亿欧元合作协议，能带领5G走向新发展吗?
小米Mix抢小米Note 2风头，这次黑科技终于不用被“黑”了
气体检测仪是如何校准的，步骤是怎样的
AL5524 IO如何扩展单片机的IO引脚数量
本项目基于物联量平台远程的视频监控项目，通过mqtt协议实现两个设备间的数据上报与订阅。通过这个项目来演示，两个mqtt设备如何互相订阅，进行消息流转。在阿里云服务器上创建2个设备，分为为设备a和设备b；设备a负责采集本地摄像头画面上传，设备b负责接收设备a上传的数据然后解析显示出来。在阿里云服务器上需要配置云产品流转，让设备a的数据上传后自动发送给设备b。这样就完成了视频画面数据的流转。不过因为阿里云的最大数据限制，每次最大发送10240字节的数据。
2.linux 下 socket 编程连接阿里云物联网平台
#define server_ip asfdda.iot-as-mqtt.cn-shanghai.aliyuncs.com//服务器ip#define server_port 1883 //端口号#define clientid aasfsaxabf.imasfas|securemode=2,signmethod=hmacsha256,timestamp=1678323607797|#define username imsfea&a1sadf8xabf#define password 15566ab496e81da728a3792ebe532fd4a3f4026a2b831df5af24da06//密文 #define set_topic /sys/a14dxabf/imagfa/thing/service/property/set //订阅 #define post_topic /sys/a14sdf8xabf/imdfea/thing/event/property/post //发布 int main(){ pthread_t id; signal(sigpipe,sig_ign);/*忽略sigpipe信号*/ signal(sigalrm,signal_func);/*闹钟信号*/ sockfd=socket(af_inet,sock_stream,0); if(sockfd==-1) { printf(网络套接字打开失败n); return 0; } /*设置发送缓冲区大小*/ int nsendbuf=40*1024;//设置为 20k if(setsockopt(sockfd,sol_socket,so_sndbuf,(const char*)&nsendbuf,sizeof(int))) { printf(setsockopt(so_sndbuf) 设置错误!n); return 0; } /*域名解析*/ struct hostent *hostent; while(1) { hostent=gethostbyname(server_ip); if(hostent==null) { printf(域名解析失败n); sleep(1); } else break; } printf(主机名:%sn,hostent->h_name); printf(协议类型:%sn,(hostent->h_addrtype == af_inet)?af_inet:af_inet6); printf(ip地址长度:%dn,hostent->h_length); char *ip; for(int i=0;hostent->h_addr_list[i];i++) { ip=inet_ntoa(*(struct in_addr *)hostent->h_addr_list[i]); printf(ip=%sn,ip); } /*连接服务器*/ struct sockaddr_in addr; addr.sin_family=af_inet;//ipv4 addr.sin_port=htons(server_port);/*端口号*/ addr.sin_addr.s_addr=inet_addr(ip);//服务器ip if(connect(sockfd, (struct sockaddr *)&addr,sizeof(struct sockaddr_in))==0) { printf(服务器连接成功n); while(1) { mqtt_init(); /*登录服务器*/ if(mqtt_connect(clientid,username,password)==0) { break; } sleep(1); printf(服务器连接中....n); } printf(连接成功rn); //订阅物联网平台数据 stat=mqtt_subscribetopic(set_topic,1,1); if(stat) { close(sockfd); printf(订阅失败rn); exit(0); } printf(订阅成功rn); /*创建线程*/ pthread_create(&id, null,pth_work_func,null); pthread_detach(id);//设置分离属性 alarm(3);//闹钟函数，时间到达会产生sigalrm信号 int a=0; while(1) { sprintf(mqtt_message,{method:thing.event.property.post,params:{image:阿里云物联网平台测试}}); mqtt_publishdata(post_topic,mqtt_message,0);//发布数据 } }}
3.云产品流转
云产品流转文档： https: //help. aliyun. com/document_detail/68677. html
3.1 什么是云产品流转
设备基于 topic 与物联网平台进行通信时， 您可以在数据流转中， 编写 sql 对 topic 中的数据进行处理， 并配置转发规则将处理后的数据转发到其他设备 topic 或阿里云其他服务。
3.2 云产品流转配置
1 .创建解析器
2.关联数据源
3.关联数据目的
4.编写解析器脚本
解析器说明文档： https: //help. aliyun. com/document_detail/270931. html
格式示例：
//通过 payload 函数， 获取设备上报的消息内容， 并按照 json 格式转换。var data = payload(json); //直接流转物模型上报数据。writeiottopic(1000, topic, data);
topic 如下：
编辑好后发布即可， 至此， 阿里物联网平台配置完成。
4.代码实现
4.1 设备 a 发送方
1 .usb 摄像头应用编程
采用 linux 下 v4l2 框架初始化 usb 摄像头， 采集图像数据。
/*摄像头初始化返回值：成功返回摄像头描述符，失败返回负数*/int video_init(struct camera *camera){ int video_fd; int i=0; /*1.打开设备节点*/ video_fd=open(video_dev,o_rdwr); if(video_fd==-1)return -1; /*2.设置摄像头格式*/ struct v4l2_format format; memset(&format,0,sizeof(format)); format.type=v4l2_buf_type_video_capture;//视频捕获格式 format.fmt.pix.width=320; format.fmt.pix.height=240; format.fmt.pix.pixelformat=v4l2_pix_fmt_yuyv;//图像数据格式yuyv if(ioctl(video_fd,vidioc_s_fmt,&format))return -2; printf(图像尺寸:%d * %dn,format.fmt.pix.width,format.fmt.pix.height); camera->image_w=format.fmt.pix.width; camera->image_h=format.fmt.pix.height; /*3.向内核请求缓冲区*/ struct v4l2_requestbuffers reqbuf; memset(&reqbuf,0,sizeof(reqbuf)); reqbuf.count=4;/*缓冲区个数*/ reqbuf.type=v4l2_buf_type_video_capture;//视频捕获格式 reqbuf.memory=v4l2_memory_mmap;/*内存映射*/ if(ioctl(video_fd,vidioc_reqbufs,&reqbuf))return -3; printf(缓冲区个数:%dn,reqbuf.count); /*4.将缓冲区映射到进程空间*/ struct v4l2_buffer quebuff; for(i=0;imamp_buff[i]=mmap(null,quebuff.length,prot_read|prot_write,map_shared,video_fd,quebuff.m.offset); printf(buff[%d]=%pn,i,camera->mamp_buff[i]); camera->mmap_size=quebuff.length; } /*5.将缓冲区添加到采集队列*/ for(i=0;i
2.图片编码处理
实时采集图像数据， 将图片数据编码为 jpg 图像格式， 再进 base64 格式编码。base64 编码是一种将二进制数据转换为 ascii 字符的方法， 它使用 64 个字符来表示任意序列的二进制数据。 base64 编码后的数据长度会比原始二进制数据略长， 但可以方便地被转换为文本格式并在网络上进行传输。
3.base64 格式编码
#include static const char * base64char = abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz0123456789+/;/*函数功能：将图片编码为base64格式形参：bindata 源图片数据 base64 编码后的数据 binlength --源文件大小返回值：返回编码后的base64数据*/char * base64_encode( const unsigned char * bindata, char * base64, int binlength ){ int i, j; unsigned char current; for ( i = 0, j = 0 ; i > 2) ; current &= (unsigned char)0x3f; base64[j++] = base64char[(int)current]; current = ( (unsigned char)(bindata[i] > 4) ) & ( (unsigned char) 0x0f ); base64[j++] = base64char[(int)current]; current = ( (unsigned char)(bindata[i+1] > 6) ) & ( (unsigned char) 0x03 ); base64[j++] = base64char[(int)current]; current = ( (unsigned char)bindata[i+2] ) & ( (unsigned char)0x3f ) ; base64[j++] = base64char[(int)current]; } base64[j] = ''; return base64;}
4. base64 格式解码
/*函数功能：base64格式数据解码形参：base64 base64格式数据 bindata 保存解码成功的图像数据返回值：成功返回解码的图像大小*/static const char * base64char = abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz0123456789+/;int base64_decode( const char * base64, unsigned char * bindata ){ int i, j; unsigned char k; unsigned char temp[4]; for ( i = 0, j = 0; base64[i] != '' ; i += 4 ) { memset( temp, 0xff, sizeof(temp) ); for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i] ) temp[0]= k; } for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i+1] ) temp[1]= k; } for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i+2] ) temp[2]= k; } for ( k = 0 ; k < 64 ; k ++ ) { if ( base64char[k] == base64[i+3] ) temp[3]= k; } bindata[j++] = ((unsigned char)(((unsigned char)(temp[0] 4)&0x03)); if ( base64[i+2] == '=' ) break; bindata[j++] = ((unsigned char)(((unsigned char)(temp[1] 2)&0x0f)); if ( base64[i+3] == '=' ) break; bindata[j++] = ((unsigned char)(((unsigned char)(temp[2]
4.3 项目效果
电力线载波报警器,power line alarm
王阳元院士：新变化的历史时期，如何抓住机遇、应对挑战？
八路全双工电力线载波对讲机,Power line interphone
暖通空调设备远程智能监控解决方案
乐视最新消息：降价促销+40亿售世茂广场+4.2亿地皮撂荒+乐视员工求职难
基于阿里云MQTT物联网平台视频监控(下)
介绍两款高性能DC-40GHz的一分二电阻式功分器
追觅、小狗、戴森吸尘器对比测评，让你了解哪款最好用
Lumia 950 XL、Surface Go、苹果 M1 MacBook 跑上 Win10X ：后者性能尤其好
stm32f407浮点运算速度
Vishay推出背面绝缘的TrenchFET功率MOSFET
CyclicBarrier 任务实践
荣耀V30 PRO的多摄协同录制功能怎么样
智能物联网企业触景无限入选2023年度高成长企业TOP100榜单
XG1590射频、中频、音频放大电路的应用
自动驾驶商业落地现状与未来发展前景
中国移动与诺基亚达成10亿欧元合作协议，能带领5G走向新发展吗?
小米Mix抢小米Note 2风头，这次黑科技终于不用被“黑”了
气体检测仪是如何校准的，步骤是怎样的
AL5524 IO如何扩展单片机的IO引脚数量