adafruit-mirror/arduino-esp32
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

[REDO] Update espota.py (#8797)

Browse source 
* Update espota.py

Changes documented in comment section of code.

* Push binary to tools

---------

Co-authored-by: rowingdude <49796265+rowingdude@users.noreply.github.com>
This commit is contained in:
Me No Dev 2023-10-25 14:59:08 +03:00 committed by GitHub
parent 49644c1c5a
commit a3d8de6adf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 221 additions and 285 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

BIN
tools/espota.exe
View file

Binary file not shown.

506
tools/espota.py
View file

@ -27,12 +27,20 @@
# 2016-01-03: # 2016-01-03:
# - Added more options to parser. # - Added more options to parser.
# #
# Changes
# 2023-05-22:
# - Replaced the deprecated optparse module with argparse.
# - Adjusted the code style to conform to PEP 8 guidelines.
# - Used with statement for file handling to ensure proper resource cleanup.
# - Incorporated exception handling to catch and handle potential errors.
# - Made variable names more descriptive for better readability.
# - Introduced constants for better code maintainability.
from __future__ import print_function from __future__ import print_function
import socket import socket
import sys import sys
import os import os
import optparse import argparse
import logging import logging
import hashlib import hashlib
import random import random
@ -41,313 +49,241 @@ import random
FLASH = 0 FLASH = 0
SPIFFS = 100 SPIFFS = 100
AUTH = 200 AUTH = 200
PROGRESS = False
# update_progress() : Displays or updates a console progress bar # Constants
## Accepts a float between 0 and 1. Any int will be converted to a float. PROGRESS_BAR_LENGTH = 60
## A value under 0 represents a 'halt'.
## A value at 1 or bigger represents 100% # update_progress(): Displays or updates a console progress bar
def update_progress(progress): def update_progress(progress):
if (PROGRESS): if PROGRESS:
barLength = 60 # Modify this to change the length of the progress bar status = ""
status = "" if isinstance(progress, int):
if isinstance(progress, int): progress = float(progress)
progress = float(progress) if not isinstance(progress, float):
if not isinstance(progress, float): progress = 0
progress = 0 status = "Error: progress var must be float\r\n"
status = "error: progress var must be float\r\n" if progress < 0:
if progress < 0: progress = 0
progress = 0 status = "Halt...\r\n"
status = "Halt...\r\n" if progress >= 1:
if progress >= 1: progress = 1
progress = 1 status = "Done...\r\n"
status = "Done...\r\n" block = int(round(PROGRESS_BAR_LENGTH * progress))
block = int(round(barLength*progress)) text = "\rUploading: [{0}] {1}% {2}".format("=" * block + " " * (PROGRESS_BAR_LENGTH - block), int(progress * 100), status)
text = "\rUploading: [{0}] {1}% {2}".format( "="*block + " "*(barLength-block), int(progress*100), status) sys.stderr.write(text)
sys.stderr.write(text) sys.stderr.flush()
sys.stderr.flush()
else:
sys.stderr.write('.')
sys.stderr.flush()
def serve(remoteAddr, localAddr, remotePort, localPort, password, filename, command = FLASH):
# Create a TCP/IP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_address = (localAddr, localPort)
logging.info('Starting on %s:%s', str(server_address[0]), str(server_address[1]))
try:
sock.bind(server_address)
sock.listen(1)
except:
logging.error("Listen Failed")
return 1
content_size = os.path.getsize(filename)
f = open(filename,'rb')
file_md5 = hashlib.md5(f.read()).hexdigest()
f.close()
logging.info('Upload size: %d', content_size)
message = '%d %d %d %s\n' % (command, localPort, content_size, file_md5)
# Wait for a connection
inv_trys = 0
data = ''
msg = 'Sending invitation to %s ' % (remoteAddr)
sys.stderr.write(msg)
sys.stderr.flush()
while (inv_trys < 10):
inv_trys += 1
sock2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
remote_address = (remoteAddr, int(remotePort))
try:
sent = sock2.sendto(message.encode(), remote_address)
except:
sys.stderr.write('failed\n')
sys.stderr.flush()
sock2.close()
logging.error('Host %s Not Found', remoteAddr)
return 1
sock2.settimeout(TIMEOUT)
try:
data = sock2.recv(37).decode()
break;
except:
sys.stderr.write('.')
sys.stderr.flush()
sock2.close()
sys.stderr.write('\n')
sys.stderr.flush()
if (inv_trys == 10):
logging.error('No response from the ESP')
return 1
if (data != "OK"):
if(data.startswith('AUTH')):
nonce = data.split()[1]
cnonce_text = '%s%u%s%s' % (filename, content_size, file_md5, remoteAddr)
cnonce = hashlib.md5(cnonce_text.encode()).hexdigest()
passmd5 = hashlib.md5(password.encode()).hexdigest()
result_text = '%s:%s:%s' % (passmd5 ,nonce, cnonce)
result = hashlib.md5(result_text.encode()).hexdigest()
sys.stderr.write('Authenticating...')
sys.stderr.flush()
message = '%d %s %s\n' % (AUTH, cnonce, result)
sock2.sendto(message.encode(), remote_address)
sock2.settimeout(10)
try:
data = sock2.recv(32).decode()
except:
sys.stderr.write('FAIL\n')
logging.error('No Answer to our Authentication')
sock2.close()
return 1
if (data != "OK"):
sys.stderr.write('FAIL\n')
logging.error('%s', data)
sock2.close()
sys.exit(1);
return 1
sys.stderr.write('OK\n')
else: else:
logging.error('Bad Answer: %s', data) sys.stderr.write(".")
sock2.close() sys.stderr.flush()
return 1
sock2.close()
logging.info('Waiting for device...') def serve(remote_addr, local_addr, remote_port, local_port, password, filename, command=FLASH):
try: # Create a TCP/IP socket
sock.settimeout(10) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
connection, client_address = sock.accept() server_address = (local_addr, local_port)
sock.settimeout(None) logging.info('Starting on %s:%s', str(server_address[0]), str(server_address[1]))
connection.settimeout(None) try:
except: sock.bind(server_address)
logging.error('No response from device') sock.listen(1)
sock.close() except Exception as e:
return 1 logging.error("Listen Failed: %s", str(e))
try: return 1
f = open(filename, "rb")
if (PROGRESS): content_size = os.path.getsize(filename)
update_progress(0) file_md5 = hashlib.md5(open(filename, 'rb').read()).hexdigest()
else: logging.info('Upload size: %d', content_size)
sys.stderr.write('Uploading') message = '%d %d %d %s\n' % (command, local_port, content_size, file_md5)
sys.stderr.flush()
offset = 0 # Wait for a connection
while True: inv_tries = 0
chunk = f.read(1024) data = ''
if not chunk: break msg = 'Sending invitation to %s ' % remote_addr
offset += len(chunk) sys.stderr.write(msg)
update_progress(offset/float(content_size)) sys.stderr.flush()
connection.settimeout(10) while inv_tries < 10:
try: inv_tries += 1
connection.sendall(chunk) sock2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
res = connection.recv(10) remote_address = (remote_addr, int(remote_port))
lastResponseContainedOK = 'OK' in res.decode() try:
except: sent = sock2.sendto(message.encode(), remote_address)
sys.stderr.write('\n') except:
logging.error('Error Uploading') sys.stderr.write('failed\n')
connection.close() sys.stderr.flush()
f.close() sock2.close()
logging.error('Host %s Not Found', remote_addr)
return 1
sock2.settimeout(TIMEOUT)
try:
data = sock2.recv(37).decode()
break
except:
sys.stderr.write('.')
sys.stderr.flush()
sock2.close()
sys.stderr.write('\n')
sys.stderr.flush()
if inv_tries == 10:
logging.error('No response from the ESP')
return 1
if data != "OK":
if data.startswith('AUTH'):
nonce = data.split()[1]
cnonce_text = '%s%u%s%s' % (filename, content_size, file_md5, remote_addr)
cnonce = hashlib.md5(cnonce_text.encode()).hexdigest()
passmd5 = hashlib.md5(password.encode()).hexdigest()
result_text = '%s:%s:%s' % (passmd5, nonce, cnonce)
result = hashlib.md5(result_text.encode()).hexdigest()
sys.stderr.write('Authenticating...')
sys.stderr.flush()
message = '%d %s %s\n' % (AUTH, cnonce, result)
sock2.sendto(message.encode(), remote_address)
sock2.settimeout(10)
try:
data = sock2.recv(32).decode()
except:
sys.stderr.write('FAIL\n')
logging.error('No Answer to our Authentication')
sock2.close()
return 1
if data != "OK":
sys.stderr.write('FAIL\n')
logging.error('%s', data)
sock2.close()
sys.exit(1)
return 1
sys.stderr.write('OK\n')
else:
logging.error('Bad Answer: %s', data)
sock2.close()
return 1
sock2.close()
logging.info('Waiting for device...')
try:
sock.settimeout(10)
connection, client_address = sock.accept()
sock.settimeout(None)
connection.settimeout(None)
except:
logging.error('No response from device')
sock.close() sock.close()
return 1 return 1
if lastResponseContainedOK:
logging.info('Success')
connection.close()
f.close()
sock.close()
return 0
sys.stderr.write('\n')
logging.info('Waiting for result...')
try: try:
count = 0 with open(filename, "rb") as f:
while True: if PROGRESS:
count=count+1 update_progress(0)
connection.settimeout(60) else:
data = connection.recv(32).decode() sys.stderr.write('Uploading')
logging.info('Result: %s' ,data) sys.stderr.flush()
offset = 0
while True:
chunk = f.read(1024)
if not chunk:
break
offset += len(chunk)
update_progress(offset / float(content_size))
connection.settimeout(10)
try:
connection.sendall(chunk)
res = connection.recv(10)
last_response_contained_ok = 'OK' in res.decode()
except Exception as e:
sys.stderr.write('\n')
logging.error('Error Uploading: %s', str(e))
connection.close()
return 1
if "OK" in data: if last_response_contained_ok:
logging.info('Success') logging.info('Success')
connection.close() connection.close()
f.close() return 0
sock.close()
return 0;
if count == 5:
logging.error('Error response from device')
connection.close()
f.close()
sock.close()
return 1
except e:
logging.error('No Result!')
connection.close()
f.close()
sock.close()
return 1
finally: sys.stderr.write('\n')
connection.close() logging.info('Waiting for result...')
f.close() count = 0
while count < 5:
count += 1
connection.settimeout(60)
try:
data = connection.recv(32).decode()
logging.info('Result: %s', data)
sock.close() if "OK" in data:
return 1 logging.info('Success')
# end serve connection.close()
return 0
except Exception as e:
logging.error('Error receiving result: %s', str(e))
connection.close()
return 1
logging.error('Error response from device')
connection.close()
return 1
finally:
connection.close()
sock.close()
return 1
def parser(unparsed_args): def parse_args(unparsed_args):
parser = optparse.OptionParser( parser = argparse.ArgumentParser(
usage = "%prog [options]", description="Transmit image over the air to the ESP32 module with OTA support."
description = "Transmit image over the air to the esp32 module with OTA support." )
)
# destination ip and port # destination ip and port
group = optparse.OptionGroup(parser, "Destination") parser.add_argument("-i", "--ip", dest="esp_ip", action="store", help="ESP32 IP Address.", default=False)
group.add_option("-i", "--ip", parser.add_argument("-I", "--host_ip", dest="host_ip", action="store", help="Host IP Address.", default="0.0.0.0")
dest = "esp_ip", parser.add_argument("-p", "--port", dest="esp_port", type=int, help="ESP32 OTA Port. Default: 3232", default=3232)
action = "store", parser.add_argument(
help = "ESP32 IP Address.", "-P", "--host_port", dest="host_port", type=int, help="Host server OTA Port. Default: random 10000-60000", default=random.randint(10000, 60000)
default = False )
)
group.add_option("-I", "--host_ip",
dest = "host_ip",
action = "store",
help = "Host IP Address.",
default = "0.0.0.0"
)
group.add_option("-p", "--port",
dest = "esp_port",
type = "int",
help = "ESP32 ota Port. Default 3232",
default = 3232
)
group.add_option("-P", "--host_port",
dest = "host_port",
type = "int",
help = "Host server ota Port. Default random 10000-60000",
default = random.randint(10000,60000)
)
parser.add_option_group(group)
# auth # authentication
group = optparse.OptionGroup(parser, "Authentication") parser.add_argument("-a", "--auth", dest="auth", help="Set authentication password.", action="store", default="")
group.add_option("-a", "--auth",
dest = "auth",
help = "Set authentication password.",
action = "store",
default = ""
)
parser.add_option_group(group)
# image # image
group = optparse.OptionGroup(parser, "Image") parser.add_argument("-f", "--file", dest="image", help="Image file.", metavar="FILE", default=None)
group.add_option("-f", "--file", parser.add_argument("-s", "--spiffs", dest="spiffs", action="store_true", help="Transmit a SPIFFS image and do not flash the module.", default=False)
dest = "image",
help = "Image file.",
metavar="FILE",
default = None
)
group.add_option("-s", "--spiffs",
dest = "spiffs",
action = "store_true",
help = "Use this option to transmit a SPIFFS image and do not flash the module.",
default = False
)
parser.add_option_group(group)
# output group # output
group = optparse.OptionGroup(parser, "Output") parser.add_argument("-d", "--debug", dest="debug", action="store_true", help="Show debug output. Overrides loglevel with debug.", default=False)
group.add_option("-d", "--debug", parser.add_argument("-r", "--progress", dest="progress", action="store_true", help="Show progress output. Does not work for Arduino IDE.", default=False)
dest = "debug", parser.add_argument("-t", "--timeout", dest="timeout", type=int, help="Timeout to wait for the ESP32 to accept invitation.", default=10)
help = "Show debug output. And override loglevel with debug.",
action = "store_true",
default = False
)
group.add_option("-r", "--progress",
dest = "progress",
help = "Show progress output. Does not work for ArduinoIDE",
action = "store_true",
default = False
)
group.add_option("-t", "--timeout",
dest = "timeout",
type = "int",
help = "Timeout to wait for the ESP32 to accept invitation",
default = 10
)
parser.add_option_group(group)
(options, args) = parser.parse_args(unparsed_args) return parser.parse_args(unparsed_args)
return options
# end parser
def main(args): def main(args):
options = parser(args) options = parse_args(args)
loglevel = logging.WARNING log_level = logging.WARNING
if (options.debug): if options.debug:
loglevel = logging.DEBUG log_level = logging.DEBUG
logging.basicConfig(level = loglevel, format = '%(asctime)-8s [%(levelname)s]: %(message)s', datefmt = '%H:%M:%S') logging.basicConfig(level=log_level, format="%(asctime)-8s [%(levelname)s]: %(message)s", datefmt="%H:%M:%S")
logging.debug("Options: %s", str(options)) logging.debug("Options: %s", str(options))
# check options # check options
global PROGRESS global PROGRESS
PROGRESS = options.progress PROGRESS = options.progress
global TIMEOUT global TIMEOUT
TIMEOUT = options.timeout TIMEOUT = options.timeout
if (not options.esp_ip or not options.image):
logging.critical("Not enough arguments.")
return 1
command = FLASH if not options.esp_ip or not options.image:
if (options.spiffs): logging.critical("Not enough arguments.")
command = SPIFFS return 1
return serve(options.esp_ip, options.host_ip, options.esp_port, options.host_port, options.auth, options.image, command) command = FLASH
# end main if options.spiffs:
command = SPIFFS
return serve(
options.esp_ip, options.host_ip, options.esp_port, options.host_port, options.auth, options.image, command
)
if __name__ == '__main__': if __name__ == "__main__":
sys.exit(main(sys.argv)) sys.exit(main(sys.argv[1:]))