From 2209046c6f45ea4f3946c7c7a78bbe23629e99ce Mon Sep 17 00:00:00 2001 From: Theo Arends <11044339+arendst@users.noreply.github.com> Date: Wed, 12 May 2021 18:34:32 +0200 Subject: [PATCH] Add MQTT OTA binary file transfer --- tasmota/xdrv_02_mqtt_1_file.ino | 89 ++++++++++++------- tools/mqtt-file/upload-ota-201.py | 142 ++++++++++++++++++++++++++++++ 2 files changed, 200 insertions(+), 31 deletions(-) create mode 100644 tools/mqtt-file/upload-ota-201.py diff --git a/tasmota/xdrv_02_mqtt_1_file.ino b/tasmota/xdrv_02_mqtt_1_file.ino index 36f0591a6..6aeb359a4 100644 --- a/tasmota/xdrv_02_mqtt_1_file.ino +++ b/tasmota/xdrv_02_mqtt_1_file.ino @@ -63,7 +63,8 @@ uint32_t FileUploadChunckSize(void) { */ const uint32_t PubSubClientHeaderSize = 5; // MQTT_MAX_HEADER_SIZE - return (((MQTT_MAX_PACKET_SIZE - PubSubClientHeaderSize - FMqtt.topic_size - FileTransferHeaderSize) / 4) * 3) -2; +// return (((MQTT_MAX_PACKET_SIZE - PubSubClientHeaderSize - FMqtt.topic_size - FileTransferHeaderSize) / 4) * 3) -2; + return MQTT_MAX_PACKET_SIZE - PubSubClientHeaderSize - FMqtt.topic_size - FileTransferHeaderSize; } void CmndFileUpload(void) { @@ -77,23 +78,33 @@ void CmndFileUpload(void) { */ const char* base64_data = nullptr; uint32_t rcv_id = 0; - char* dataBuf = (char*)XdrvMailbox.data; - if (strlen(dataBuf) > 8) { // Workaround exception if empty JSON like {} - Needs checks - JsonParser parser((char*) dataBuf); - JsonParserObject root = parser.getRootObject(); - if (root) { - JsonParserToken val = root[PSTR("ID")]; - if (val) { rcv_id = val.getUInt(); } - val = root[PSTR("TYPE")]; - if (val) { FMqtt.file_type = val.getUInt(); } - val = root[PSTR("SIZE")]; - if (val) { FMqtt.file_size = val.getUInt(); } - val = root[PSTR("MD5")]; - if (val) { FMqtt.file_md5 = val.getStr(); } - val = root[PSTR("DATA")]; - if (val) { base64_data = val.getStr(); } + + bool binary_data = false; + if (XdrvMailbox.index > 199) { // Check for raw data + XdrvMailbox.index -= 200; + binary_data = true; + } + + if (!binary_data) { + if (strlen(dataBuf) > 8) { // Workaround exception if empty JSON like {} - Needs checks + JsonParser parser((char*) dataBuf); + JsonParserObject root = parser.getRootObject(); + if (root) { + JsonParserToken val = root[PSTR("ID")]; + if (val) { rcv_id = val.getUInt(); } + val = root[PSTR("TYPE")]; + if (val) { FMqtt.file_type = val.getUInt(); } + val = root[PSTR("SIZE")]; + if (val) { FMqtt.file_size = val.getUInt(); } + val = root[PSTR("MD5")]; + if (val) { FMqtt.file_md5 = val.getStr(); } + val = root[PSTR("DATA")]; + if (val) { base64_data = val.getStr(); } + } } + } else { + rcv_id = FMqtt.file_id; } if ((0 == FMqtt.file_id) && (rcv_id > 0) && (FMqtt.file_size > 0) && (FMqtt.file_type > 0)) { @@ -139,24 +150,36 @@ void CmndFileUpload(void) { } if (FMqtt.file_buffer) { - if ((FMqtt.file_pos < FMqtt.file_size) && base64_data) { + if ((FMqtt.file_pos < FMqtt.file_size) && (binary_data || base64_data)) { // Save upload into buffer - Handle possible buffer overflows - uint32_t rcvd_bytes = decode_base64_length((unsigned char*)base64_data); - unsigned char decode_output[rcvd_bytes]; - decode_base64((unsigned char*)base64_data, (unsigned char*)decode_output); - uint32_t bytes_left = FMqtt.file_size - FMqtt.file_pos; - uint32_t read_bytes = (bytes_left < rcvd_bytes) ? bytes_left : rcvd_bytes; - FMqtt.md5.add(decode_output, read_bytes); - - if (UPL_TASMOTA == FMqtt.file_type) { - Update.write(decode_output, read_bytes); - } else { - uint8_t* buffer = FMqtt.file_buffer + FMqtt.file_pos; - memcpy(buffer, decode_output, read_bytes); + unsigned char* raw_data = (unsigned char*)XdrvMailbox.data; + uint32_t rcvd_bytes = XdrvMailbox.data_len; + if (!binary_data) { + raw_data = (unsigned char*)malloc(XdrvMailbox.data_len); + if (raw_data) { + rcvd_bytes = decode_base64((unsigned char*)base64_data, (unsigned char*)raw_data); + } } - FMqtt.file_pos += read_bytes; + if (raw_data) { + uint32_t bytes_left = FMqtt.file_size - FMqtt.file_pos; + uint32_t read_bytes = (bytes_left < rcvd_bytes) ? bytes_left : rcvd_bytes; + FMqtt.md5.add(raw_data, read_bytes); + + if (UPL_TASMOTA == FMqtt.file_type) { + Update.write(raw_data, read_bytes); + } else { + uint8_t* buffer = FMqtt.file_buffer + FMqtt.file_pos; + memcpy(buffer, raw_data, read_bytes); + } + + if (!binary_data) { + free(raw_data); + } + + FMqtt.file_pos += read_bytes; + } if ((FMqtt.file_pos > rcvd_bytes) && ((FMqtt.file_pos % 102400) <= rcvd_bytes)) { TasmotaGlobal.masterlog_level = LOG_LEVEL_NONE; // Enable logging @@ -168,8 +191,12 @@ void CmndFileUpload(void) { if ((FMqtt.file_pos < FMqtt.file_size) || (FMqtt.file_md5.length() != 32)) { TasmotaGlobal.masterlog_level = LOG_LEVEL_DEBUG_MORE; // Hide upload data logging + uint32_t chunk_size = FileUploadChunckSize(); + if (!binary_data) { + chunk_size = ((chunk_size / 4) * 3) -2; // Calculate base64 chunk size + } // {"Id":116,"MaxSize":"765"} - Response_P(PSTR("{\"Id\":%d,\"MaxSize\":%d}"), FMqtt.file_id, FileUploadChunckSize()); + Response_P(PSTR("{\"Id\":%d,\"MaxSize\":%d}"), FMqtt.file_id, chunk_size); } else { FMqtt.md5.calculate(); if (strcasecmp(FMqtt.file_md5.c_str(), FMqtt.md5.toString().c_str())) { diff --git a/tools/mqtt-file/upload-ota-201.py b/tools/mqtt-file/upload-ota-201.py new file mode 100644 index 000000000..b9af8e214 --- /dev/null +++ b/tools/mqtt-file/upload-ota-201.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +""" + upload-ota-201.py - Upload Tasmota firmware file + + Copyright (C) 2021 Theo Arends + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program. If not, see . + +Requirements: + - Python 3.x and Pip: + sudo apt-get install python3 python3-pip + pip3 install paho-mqtt json + +Instructions: + Edit file and change parameters in User Configuration Section + + Then execute command upload-ota-201.py + +""" + +import paho.mqtt.client as mqtt +import time +import base64 +import hashlib +import json + +# **** Start of User Configuration Section + +broker = "domus1" # MQTT broker ip address or name +broker_port = 1883 # MQTT broker port + +mytopic = "demo" # Tasmota MQTT topic +#myfile = "../../build_output/firmware/tasmota32.bin" # Tasmota esp32 firmware file name +myfile = "../../build_output/firmware/tasmota.bin.gz" # Tasmota esp8266 firmware file name +myfiletype = 1 # Tasmota firmware file type + +# **** End of User Configuration Section + +# Derive fulltopic from broker LWT message +mypublish = "cmnd/"+mytopic+"/fileupload" +mysubscribe = "stat/"+mytopic+"/FILEUPLOAD" # Case sensitive + +Ack_flag = False + +file_id = 114 # Even id between 2 and 254 +file_chunk_size = 700 # Default Tasmota MQTT max message size + +# The callback for when mysubscribe message is received +def on_message(client, userdata, msg): + global Ack_flag + global file_chunk_size + + rcv_code = "" + rcv_id = 0 + +# print("Received message =",str(msg.payload.decode("utf-8"))) + + root = json.loads(msg.payload.decode("utf-8")) + if "Command" in root: rcv_code = root["Command"] + if rcv_code == "Error": + print("Error: Command error") + return + + if "Id" in root: rcv_id = root["Id"] + if rcv_id == file_id: + if "MaxSize" in root: file_chunk_size = root["MaxSize"] + + Ack_flag = False + +def wait_for_ack(): + global Ack_flag + timeout = 100 + while Ack_flag and timeout > 0: + time.sleep(0.01) + timeout = timeout -1 + + if Ack_flag: + print("Error: Ack timeout") + + return Ack_flag + +client = mqtt.Client() +client.on_message = on_message +client.connect(broker, broker_port) +client.loop_start() # Start loop to process received messages +client.subscribe(mysubscribe) + +time_start = time.time() +print("Uploading file "+myfile+" to "+mytopic+" ...") + +fo = open(myfile,"rb") +fo.seek(0, 2) # os.SEEK_END +file_size = fo.tell() +fo.seek(0, 0) # os.SEEK_SET + +client.publish(mypublish, "{\"File\":\""+myfile+"\",\"Id\":"+str("%3d"%file_id)+",\"Type\":"+str(myfiletype)+",\"Size\":"+str(file_size)+"}") +Ack_flag = True + +out_hash_md5 = hashlib.md5() + +Run_flag = True +while Run_flag: + if wait_for_ack(): # We use Ack here + Run_flag = False + + else: + chunk = fo.read(file_chunk_size) + if chunk: + out_hash_md5.update(chunk) # Update hash + +# base64_encoded_data = base64.b64encode(chunk) +# base64_data = base64_encoded_data.decode('utf-8') + # Message length used by Tasmota (FileTransferHeaderSize) +# client.publish(mypublish, "{\"Id\":"+str("%3d"%file_id)+",\"Data\":\""+base64_data+"\"}") + client.publish(mypublish+"201", chunk) + Ack_flag = True + + else: + md5_hash = out_hash_md5.hexdigest() + client.publish(mypublish, "{\"Id\":"+str("%3d"%file_id)+",\"Md5\":\""+md5_hash+"\"}") + Run_flag = False + +fo.close() + +time_taken = time.time() - time_start +print("Done in "+str("%.2f"%time_taken)+" seconds") + +client.disconnect() # Disconnect +client.loop_stop() # Stop loop