# cd demo/ceshi-1/ # python ceshi.py import serial import time import struct import json import paho.mqtt.client as mqtt from api import add from db.models.log_data_model import LOT_DATA def hex_to_float(hex_str): hex_int = int(hex_str, 16) return struct.unpack('!f', struct.pack('!I', hex_int))[0] def print_json(data): print(json.dumps(data, sort_keys=True, indent=4, separators=(', ', ': '), ensure_ascii=False)) def run_with_client(client): temp_send = '06 03 01 F4 00 02 85 B2 ' # 温湿度查询指令 co2_send = '06 03 01 F7 00 02 75 B2 ' # 二氧化碳查询指令 pressure_send = '06 03 01 F9 00 02 14 71 ' # 气压查询指令 sun_send = '06 03 01 FA 00 02 E4 71 ' # 光照查询指令 soil_send = '02 03 00 00 00 04 44 3A' # 土壤查询指令 danlinjia_send = '02 03 00 04 00 03 44 39' # 氮磷钾查询指令 rainfall_send = '03 03 00 00 00 01 85 E8' # 雨量查询指令 windspeed_send = '04 03 00 00 00 02 C4 5E' # 风速查询指令 winddirection_send = '05 03 00 00 00 02 C5 8F' # 风向查询指令 ser = serial.Serial("/dev/ttyS2", 9600) # 发送的数据转为2进制b'\x01\x03\x00\x00\x00\x02\xc4\x0b' temp_send = bytes.fromhex(temp_send) co2_send = bytes.fromhex(co2_send) pressure_send = bytes.fromhex(pressure_send) sun_send = bytes.fromhex(sun_send) soil_send = bytes.fromhex(soil_send) danlinjia_send = bytes.fromhex(danlinjia_send) rainfall_send = bytes.fromhex(rainfall_send) windspeed_send = bytes.fromhex(windspeed_send) winddirection_send = bytes.fromhex(winddirection_send) while True: # if ser.is_open: ser.write(temp_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 tempbuffer_data = ser.in_waiting if tempbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(tempbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 airtemp_data = int(return_data_hex[6:10], 16) / 10 airhumi_data = int(return_data_hex[10:14], 16) / 10 time.sleep(5) ser.write(co2_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 co2buffer_data = ser.in_waiting if co2buffer_data: return_data = ser.read(co2buffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 co2_data = int(return_data_hex[6:10], 16) time.sleep(5) ser.write(pressure_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 pressurebuffer_data = ser.in_waiting if pressurebuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(pressurebuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 pressure_data = int(return_data_hex[6:10], 16) / 10 time.sleep(5) ser.write(sun_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 sunbuffer_data = ser.in_waiting if sunbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(sunbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) sun_data = int(return_data_hex[6:14], 16) time.sleep(5) # print('send soil directives') ser.write(soil_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 soilbuffer_data = ser.in_waiting # print(buffer_data, 'buffer_data') if soilbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(soilbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 humidity_data = int(return_data_hex[6:10], 16) / 10 temperature_data = int(return_data_hex[10:14], 16) / 10 electrical_data = int(return_data_hex[14:18], 16) / 10 PH_data = int(return_data_hex[18:22], 16) / 10 time.sleep(5) # print('send danlinjia directives') ser.write(danlinjia_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 danlinjiabuffer_data = ser.in_waiting # print(buffer_data, 'buffer_data') if danlinjiabuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(danlinjiabuffer_data) # print('返回的数据2进制:', return_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # # 对返回的数据进行解析，获取温度和湿度数据 dan_data = int(return_data_hex[6:10], 16) / 10 lin_data = int(return_data_hex[10:14], 16) / 10 jia_data = int(return_data_hex[14:18], 16) / 10 time.sleep(5) # print('send rainfall directives') ser.write(rainfall_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 rainfallbuffer_data = ser.in_waiting # print(buffer_data, 'buffer_data') if rainfallbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(rainfallbuffer_data) # print('返回的数据2进制:', return_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # print('返回的数据转换为16进制:', return_data_hex) # # 对返回的数据进行解析，获取温度和湿度数据 # print("当前雨量值为：", int(return_data_hex[6:10], 16)/10)#单位mm rainfall_data = int(return_data_hex[6:10], 16) / 10 time.sleep(5) ser.write(windspeed_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 windspeedbuffer_data = ser.in_waiting # print(buffer_data, 'buffer_data') if windspeedbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(windspeedbuffer_data) # print('返回的数据2进制:', return_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # print('返回的数据转换为16进制:', return_data_hex) # print("当前风速为：", int(return_data_hex[6:10], 16)/10)#单位mm speedwind_data = int(return_data_hex[6:10], 16) / 10 time.sleep(5) # print('send winddirection directives') ser.write(winddirection_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 winddirectionbuffer_data = ser.in_waiting # print(buffer_data, 'buffer_data') if winddirectionbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(winddirectionbuffer_data) # print('返回的数据2进制:', return_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # print('返回的数据转换为16进制:', return_data_hex) # print("当前风向为：", int(return_data_hex[10:14], 16))#单位mm winddirection_data = int(return_data_hex[10:14], 16) time.sleep(5) # print('{"name":"%d","name1":"%d"}', 123,456) # data = [{'ngvhgv': airtemp_data}, {'nvjgvjvj':airhumi_data}] data = {'ambient_temperature': airtemp_data, 'ambient_humidity': airhumi_data, 'carbon_dioxide': co2_data, 'ambient_air_pressure': pressure_data, 'ambient_lighting': sun_data, 'soil_moisture': humidity_data, 'soil_temperature': temperature_data, 'soil_conductivity': electrical_data, 'soil_PH': PH_data, 'soil_potassium_phosphate_nitrogen': dan_data, 'soil_potassium_phosphate_phosphorus': lin_data, 'soil_potassium_phosphate_potassium': jia_data, 'rainfall': rainfall_data, 'wind_speed': speedwind_data, 'wind_direction': winddirection_data, 'create_time': int(time.time()) } t = LOT_DATA(**data) # TODO 判断数据是否正常 # 发送给服务器 client.publish('demo', payload=json.dumps(data, ensure_ascii=False), qos=0) add(t) def run_no_client(): temp_send = '06 03 01 F4 00 02 85 B2 ' # 温湿度查询指令 co2_send = '06 03 01 F7 00 02 75 B2 ' # 二氧化碳查询指令 pressure_send = '06 03 01 F9 00 02 14 71 ' # 气压查询指令 sun_send = '06 03 01 FA 00 02 E4 71 ' # 光照查询指令 soil_send = '02 03 00 00 00 04 44 3A' # 土壤查询指令 danlinjia_send = '02 03 00 04 00 03 44 39' # 氮磷钾查询指令 rainfall_send = '03 03 00 00 00 01 85 E8' # 雨量查询指令 windspeed_send = '04 03 00 00 00 02 C4 5E' # 风速查询指令 winddirection_send = '05 03 00 00 00 02 C5 8F' # 风向查询指令 ser = serial.Serial("/dev/ttyS2", 9600) # 发送的数据转为2进制b'\x01\x03\x00\x00\x00\x02\xc4\x0b' temp_send = bytes.fromhex(temp_send) co2_send = bytes.fromhex(co2_send) pressure_send = bytes.fromhex(pressure_send) sun_send = bytes.fromhex(sun_send) soil_send = bytes.fromhex(soil_send) danlinjia_send = bytes.fromhex(danlinjia_send) rainfall_send = bytes.fromhex(rainfall_send) windspeed_send = bytes.fromhex(windspeed_send) winddirection_send = bytes.fromhex(winddirection_send) while True: ser.write(temp_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 tempbuffer_data = ser.in_waiting if tempbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(tempbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 airtemp_data = int(return_data_hex[6:10], 16) / 10 airhumi_data = int(return_data_hex[10:14], 16) / 10 time.sleep(5) ser.write(co2_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 co2buffer_data = ser.in_waiting if co2buffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(co2buffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 co2_data = int(return_data_hex[6:10], 16) time.sleep(5) ser.write(pressure_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 pressurebuffer_data = ser.in_waiting if pressurebuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(pressurebuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 pressure_data = int(return_data_hex[6:10], 16) / 10 time.sleep(5) ser.write(sun_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 sunbuffer_data = ser.in_waiting if sunbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(sunbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) sun_data = int(return_data_hex[6:14], 16) time.sleep(5) ser.write(soil_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 soilbuffer_data = ser.in_waiting if soilbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(soilbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # 对返回的数据进行解析，获取温度和湿度数据 humidity_data = int(return_data_hex[6:10], 16) / 10 temperature_data = int(return_data_hex[10:14], 16) / 10 electrical_data = int(return_data_hex[14:18], 16) / 10 PH_data = int(return_data_hex[18:22], 16) / 10 time.sleep(5) ser.write(danlinjia_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 danlinjiabuffer_data = ser.in_waiting if danlinjiabuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(danlinjiabuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # # 对返回的数据进行解析，获取温度和湿度数据 dan_data = int(return_data_hex[6:10], 16) / 10 lin_data = int(return_data_hex[10:14], 16) / 10 jia_data = int(return_data_hex[14:18], 16) / 10 time.sleep(5) ser.write(rainfall_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 rainfallbuffer_data = ser.in_waiting if rainfallbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(rainfallbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) # # 对返回的数据进行解析，获取温度和湿度数据 rainfall_data = int(return_data_hex[6:10], 16) / 10 time.sleep(5) ser.write(windspeed_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 windspeedbuffer_data = ser.in_waiting if windspeedbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(windspeedbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) speedwind_data = int(return_data_hex[6:10], 16) / 10 time.sleep(5) ser.write(winddirection_send) time.sleep(1) # 获取返回的缓冲data,获取的是buffer_data的长度 9 winddirectionbuffer_data = ser.in_waiting if winddirectionbuffer_data: # 返回的数据为2进制：b'\x01\x03\x04\x01\x08\x022\xfa\xb8' return_data = ser.read(winddirectionbuffer_data) # 二进制转换为16进制：010304010802307b79 return_data_hex = str(return_data.hex()) winddirection_data = int(return_data_hex[10:14], 16) time.sleep(5) data = {'ambient_temperature': airtemp_data, 'ambient_humidity': airhumi_data, 'carbon_dioxide': co2_data, 'ambient_air_pressure': pressure_data, 'ambient_lighting': sun_data, 'soil_moisture': humidity_data, 'soil_temperature': temperature_data, 'soil_conductivity': electrical_data, 'soil_PH': PH_data, 'soil_potassium_phosphate_nitrogen': dan_data, 'soil_potassium_phosphate_phosphorus': lin_data, 'soil_potassium_phosphate_potassium': jia_data, 'rainfall': rainfall_data, 'wind_speed': speedwind_data, 'wind_direction': winddirection_data} data = LOT_DATA(**data, create_time=int(time.time())) # TODO 判断数据是否正常 add(data) class UPLOAD: def __init__(self): self.times = 3 self.client = mqtt.Client(transport="websockets") self.client.username_pw_set("demo", "123456") # Specify callback function self.client.on_connect = self.on_connect # Establish a connection self.client.connect('ceshi-mqtt.lihaink.cn', 8083) # Publish a message self.client.loop_forever() def on_connect(self, client, userdata, flags, rc): from threading import Thread if rc == 0: print("连接成功，执行数据推送和本地存储") nt1 = Thread(target=self.t1) nt1.start() else: if self.times != 0: self.times -= 1 client.reconnect() else: print("3次失败，执行本地存储") nt2 = Thread(target=self.t2) nt2.start() def t2(self): run_no_client() def t1(self): run_with_client(self.client)