协议为：
需要将图片数据封装成多个消息进行传输，每个消息传输的数据字节数为1400byte。
消息(mqtt payload) 格式：web服务器-------->base：
反馈：base---------> web服务器:
如果web服务器发送完一个“数据传输消息”后，5s内没有收到mqtt“反馈消息”或者收到的反馈中显示“数据包不完整”，则重发该“数据传输消息”。
程序流程图
根据上面的协议，可以得到如下的流程图：
代码如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
# encoding:utf-8
from flask import flask, jsonify
from flask_restful import api, resource, reqparse
from pil import image
from io import bytesio
import requests
import os, logging, time
import paho.mqtt.client as mqtt
import struct
from flask_cors import *
# 日志配置信息
logging.basicconfig(
level=logging.info,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s （runing by %(funcname)s',
)
class mqtt(object):
def __init__(self, img_data, size):
self.mqtthost = '*******'
self.mqttport = "******"
# 订阅和发送的主题
self.topic_from_base = 'mqtttestsub'
self.topic_to_base = 'mqtttestpub'
self.client_id = time.strftime('%y%m%d%h%m%s', time.localtime(time.time()))
self.client = mqtt.client(self.client_id)
# 完成链接后的回掉函数
self.client.on_connect = self.on_connect
# 图片大小
self.size = size
# 用于跳出死循环，结束任务
self.finished = none
# 包的编号
self.index = 0
# 将收到的图片数据按大小分成列表
self.image_data_list = [img_data[x:x + 1400] for x in range(0, self.size, 1400)]
# 记录发布后的数据，用于监控时延
self.pub_time = 0
self.header_to_base = 0xffffeeee
self.header_from_base = 0xeeeeffff
# 功能标识
self.function_begin = 0x01
self.function_doing = 0x02
self.function_finished = 0x03
# 包的完整和非完整状态
self.whole_package = 0x01
self.bad_package = 0x00
# 头信息的格式,小端模式
self.format_to_base = "<lbhh"
self.format_from_base = "<lbhb"
# 如果重发包时，用于检查是否重发第一个包
self.first = true
# 如果重发包时，用于检查是否重发最后一个包
self.last = false
self.begin_data = 'image.jpg;' + str(self.size)
# 链接mqtt服务器函数
def on_mqtt_connect(self):
self.client.connect(self.mqtthost, self.mqttport, 60)
self.client.loop_start()
# 链接完成后的回调函数
def on_connect(self, client, userdata, flags, rc):
logging.info("+++ connected with result code {} +++".format(str(rc)))
self.client.subscribe(self.topic_from_base)
# 订阅函数
def subscribe(self):
self.client.subscribe(self.topic_from_base, 1)
# 消息到来处理函数
self.client.on_message = self.on_message
# 接收到信息后的回调函数
def on_message(self, client, userdata, msg):
# 如果接受第一个包则不需要重发第一个
self.first = false
# 将接受到的包进行解压，得到一个元组
base_tuple = struct.unpack(self.format_from_base, msg.payload)
logging.info("+++ imagedata's letgth is {}, base_tupe is {} +++".format(self.size, base_tuple))
logging.info("+++ package_number is {}, package_status_from_base is {} +++"
.format(base_tuple[2], base_tuple[3]))
# 检查接受到信息的头部是否正确
if base_tuple[0] == self.header_from_base:
logging.info("+++ function_from_base is {} +++".format(base_tuple[1]))
# 是否完成传输，如果完成则退出
if base_tuple[1] == self.function_finished:
logging.info("+++ finish work +++")
self.finished = 1
self.client.disconnect()
else:
# 是否是最后一个包
if self.index == len(self.image_data_list) - 1:
self.publish('finished', self.function_finished)
self.last = true
logging.info("+++ finished_data_to_base is finished+++")
else:
# 如果接收到的包不是 0x03则进行传送数据
if base_tuple[1] == self.function_begin or base_tuple[1] == self.function_doing:
logging.info("+++ package_number is {}, package_status_from_base is {} +++"
.format(base_tuple[2],base_tuple[3]))
# 如果数据的反馈中，包的状态是1则继续发下一个包
if base_tuple[3] == self.whole_package:
self.publish(self.index, self.function_doing)
logging.info("+++ data_to_base is finished+++")
self.index += 1
# 如果数据的反馈中，包的状态是0则重发数据包
elif base_tuple[3] == self.bad_package:
re_package_number = base_tuple[2]
self.publish(re_package_number-1, self.function_doing)
logging.info("+++ re_data_to_base is finished+++")
else:
logging.info("+++ package_status_from_base is not 0 or 1 +++")
self.client.disconnect()
else:
logging.info("+++ function_identifier is illegal +++")
self.client.disconnect()
else:
logging.info("+++ header_from_base is illegal +++")
self.client.disconnect()
# 数据发送函数
def publish(self, index, fuc):
# 看是否是最后一个包
if index == 'finished':
length = 0
package_number = 0
data = b''
else:
length = len(self.image_data_list[index])
package_number = index
data = self.image_data_list[index]
# 打包数据头信息
buffer = struct.pack(
self.format_to_base,
self.header_to_base,
fuc,
package_number,
length
)
to_base_data = buffer + data
# mqtt发送
self.client.publish(
self.topic_to_base,
to_base_data
)
self.pub_time = time.time()
# 发送第一个包函数
def publish_begin(self):
buffer = struct.pack(
self.format_to_base,
self.header_to_base,
self.function_begin,
0,
len(self.begin_data.encode('utf-8')),
)
begin_data = buffer + self.begin_data.encode('utf-8')
self.client.publish(self.topic_to_base, begin_data)
# 控制函数
def control(self):
self.on_mqtt_connect()
self.publish_begin()
begin_time = time.time()
self.pub_time = time.time()
self.subscribe()
while true:
time.sleep(1)
# 超过5秒重传
date = time.time() - self.pub_time
if date > 5:
# 是否重传第一个包
if self.first == true:
self.publish_begin()
logging.info('+++ this is timeout first_data +++')
# 是否重传最后一个包
elif self.last == true:
self.publish('finished', self.function_finished)
logging.info('+++ this is timeout last_data +++')
else:
self.publish(self.index-1, self.function_doing)
logging.info('+++ this is timeout middle_data +++')
if self.finished == 1:
logging.info('+++ all works is finished+++')
break
print(str(time.time()-begin_time) + 'begin_time - end_time')
app = flask(__name__)
api = api(app)
cors(app, supports_credentials=true)
# 接受参数
parser = reqparse.requestparser()
parser.add_argument('url', help='mqttimage url', location='args', type=str)
class getimage(resource):
# 得到参数并从图床下载到本地
def get(self):
args = parser.parse_args()
url = args.get('url')
response = requests.get(url)
# 获取图片
image = image.open(bytesio(response.content))
# 存取图片
add = os.path.join(os.path.abspath(''), 'image.jpg')
image.save(add)
# 得到图片大小
size = os.path.getsize(add)
f = open(add, 'rb')
imagedata = f.read()
f.close()
# 进行mqtt传输
mqtt = mqtt(imagedata, size)
mqtt.control()
# 删除文件
os.remove(add)
logging.info('*** the result of control is {} ***'.format(1))
return jsonify({
"imagedata": 1
})
api.add_resource(getimage, '/image')
if __name__ == '__main__':
app.run(debug=true, host='0.0.0.0')
|
总结
以上所述是小编给大家介绍的python使用mqtt给硬件传输图片的实现方法,希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。在此也非常感谢大家对服务器之家网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！原文链接：https://juejin.im/post/5ccc4bd1f265da037c7cf4a5