Remove OTA feature #1738

This commit is contained in:
proddy
2024-05-04 14:19:19 +02:00
parent 6537abedb9
commit dabb958f06
44 changed files with 291 additions and 824 deletions

View File

@@ -1,343 +0,0 @@
#!/usr/bin/env python3
#
# Original espota.py by Ivan Grokhotkov:
# https://gist.github.com/igrr/d35ab8446922179dc58c
#
# Modified since 2015-09-18 from Pascal Gollor (https://github.com/pgollor)
# Modified since 2015-11-09 from Hristo Gochkov (https://github.com/me-no-dev)
# Modified since 2016-01-03 from Matthew O'Gorman (https://githumb.com/mogorman)
#
# This script will push an OTA update to the ESP
# use it like: python3 espota.py -i <ESP_IP_address> -I <Host_IP_address> -p <ESP_port> -P <Host_port> [-a password] -f <sketch.bin>
# Or to upload SPIFFS image:
# python3 espota.py -i <ESP_IP_address> -I <Host_IP_address> -p <ESP_port> -P <HOST_port> [-a password] -s -f <spiffs.bin>
#
# Changes
# 2015-09-18:
# - Add option parser.
# - Add logging.
# - Send command to controller to differ between flashing and transmitting SPIFFS image.
#
# Changes
# 2015-11-09:
# - Added digest authentication
# - Enhanced error tracking and reporting
#
# Changes
# 2016-01-03:
# - Added more options to parser.
#
from __future__ import print_function
import socket
import sys
import os
import optparse
import logging
import hashlib
import random
# Commands
FLASH = 0
SPIFFS = 100
AUTH = 200
PROGRESS = False
# update_progress() : Displays or updates a console progress bar
## Accepts a float between 0 and 1. Any int will be converted to a float.
## A value under 0 represents a 'halt'.
## A value at 1 or bigger represents 100%
def update_progress(progress):
if (PROGRESS):
barLength = 60 # Modify this to change the length of the progress bar
status = ""
if isinstance(progress, int):
progress = float(progress)
if not isinstance(progress, float):
progress = 0
status = "error: progress var must be float\r\n"
if progress < 0:
progress = 0
status = "Halt...\r\n"
if progress >= 1:
progress = 1
status = "Done...\r\n"
block = int(round(barLength*progress))
text = "\rUploading: [{0}] {1}% {2}".format( "="*block + " "*(barLength-block), int(progress*100), status)
sys.stderr.write(text)
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
# Check whether Signed Update is used.
if ( os.path.isfile(filename + '.signed') ):
filename = filename + '.signed'
file_check_msg = 'Detected Signed Update. %s will be uploaded instead.' % (filename)
sys.stderr.write(file_check_msg + '\n')
sys.stderr.flush()
logging.info(file_check_msg)
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
logging.info('Sending invitation to: %s', remoteAddr)
sock2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
remote_address = (remoteAddr, int(remotePort))
sent = sock2.sendto(message.encode(), remote_address)
sock2.settimeout(10)
try:
data = sock2.recv(128).decode()
except:
logging.error('No Answer')
sock2.close()
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:
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()
return 1
received_ok = False
try:
f = open(filename, "rb")
if (PROGRESS):
update_progress(0)
else:
sys.stderr.write('Uploading')
sys.stderr.flush()
offset = 0
while True:
chunk = f.read(1460)
if not chunk: break
offset += len(chunk)
update_progress(offset/float(content_size))
connection.settimeout(10)
try:
connection.sendall(chunk)
if connection.recv(32).decode().find('O') >= 0:
# connection will receive only digits or 'OK'
received_ok = True
except:
sys.stderr.write('\n')
logging.error('Error Uploading')
connection.close()
f.close()
sock.close()
return 1
sys.stderr.write('\n')
logging.info('Waiting for result...')
# libraries/ArduinoOTA/ArduinoOTA.cpp L311 L320
# only sends digits or 'OK'. We must not not close
# the connection before receiving the 'O' of 'OK'
try:
connection.settimeout(60)
received_ok = False
received_error = False
while not (received_ok or received_error):
reply = connection.recv(64).decode()
# Look for either the "E" in ERROR or the "O" in OK response
# Check for "E" first, since both strings contain "O"
if reply.find('E') >= 0:
sys.stderr.write('\n')
logging.error('%s', reply)
received_error = True
elif reply.find('O') >= 0:
logging.info('Result: OK')
received_ok = True
connection.close()
f.close()
sock.close()
if received_ok:
return 0
return 1
except:
logging.error('No Result!')
connection.close()
f.close()
sock.close()
return 1
finally:
connection.close()
f.close()
sock.close()
return 1
# end serve
def parser(unparsed_args):
parser = optparse.OptionParser(
usage = "%prog [options]",
description = "Transmit image over the air to the esp8266 module with OTA support."
)
# destination ip and port
group = optparse.OptionGroup(parser, "Destination")
group.add_option("-i", "--ip",
dest = "esp_ip",
action = "store",
help = "ESP8266 IP Address.",
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 = "ESP8266 ota Port. Default 8266",
default = 8266
)
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
group = optparse.OptionGroup(parser, "Authentication")
group.add_option("-a", "--auth",
dest = "auth",
help = "Set authentication password.",
action = "store",
default = ""
)
parser.add_option_group(group)
# image
group = optparse.OptionGroup(parser, "Image")
group.add_option("-f", "--file",
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
group = optparse.OptionGroup(parser, "Output")
group.add_option("-d", "--debug",
dest = "debug",
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
)
parser.add_option_group(group)
(options, args) = parser.parse_args(unparsed_args)
return options
# end parser
def main(args):
# get options
options = parser(args)
# adapt log level
loglevel = logging.WARNING
if (options.debug):
loglevel = logging.DEBUG
# end if
# logging
logging.basicConfig(level = loglevel, format = '%(asctime)-8s [%(levelname)s]: %(message)s', datefmt = '%H:%M:%S')
logging.debug("Options: %s", str(options))
# check options
global PROGRESS
PROGRESS = options.progress
if (not options.esp_ip or not options.image):
logging.critical("Not enough arguments.")
return 1
# end if
command = FLASH
if (options.spiffs):
command = SPIFFS
# end if
return serve(options.esp_ip, options.host_ip, options.esp_port, options.host_port, options.auth, options.image, command)
# end main
if __name__ == '__main__':
sys.exit(main(sys.argv))
# end if

6
scripts/requirements.txt Normal file
View File

@@ -0,0 +1,6 @@
requests
termcolor
requests_toolbelt
tqdm

130
scripts/upload.py Normal file
View File

@@ -0,0 +1,130 @@
# Modified from https://github.com/ayushsharma82/ElegantOTA
# This is called during the platformIO upload process
# To use create a pio_local.ini file in the project root and add the following:
# [env]
# upload_protocol = custom
# custom_emsesp_ip = 10.10.10.173
# custom_username = admin
# custom_password = admin
#
# and
# extra_scripts = scripts/upload.py
import requests
import hashlib
from urllib.parse import urlparse
import time
Import("env")
try:
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from tqdm import tqdm
from termcolor import cprint
except ImportError:
env.Execute("$PYTHONEXE -m pip install requests_toolbelt")
env.Execute("$PYTHONEXE -m pip install tqdm")
env.Execute("$PYTHONEXE -m pip install termcolor")
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from tqdm import tqdm
from termcolor import cprint
def print_success(x): return cprint(x, 'green')
def print_fail(x): return cprint(x, 'red')
def on_upload(source, target, env):
# first check authentication
try:
username = env.GetProjectOption('custom_username')
password = env.GetProjectOption('custom_password')
except:
print('No authentication settings specified. Please, add these to your pio_local.ini file: \n\ncustom_username=username\ncustom_password=password\n')
return
emsesp_url = "http://" + env.GetProjectOption('custom_emsesp_ip')
parsed_url = urlparse(emsesp_url)
host_ip = parsed_url.netloc
signon_url = f"{emsesp_url}/rest/signIn"
signon_headers = {
'Host': host_ip,
'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/118.0',
'Accept': '*/*',
'Accept-Language': 'de,en-US;q=0.7,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Referer': f'{emsesp_url}',
'Content-Type': 'application/json',
'Connection': 'keep-alive'
}
username_password = {
"username": username,
"password": password
}
response = requests.post(signon_url, json=username_password, headers=signon_headers, auth=None)
if response.status_code != 200:
print_fail("Authentication failed (code " + str(response.status_code) + ")")
return
print_success("Authentication successful")
access_token = response.json().get('access_token')
# start the upload
firmware_path = str(source[0])
with open(firmware_path, 'rb') as firmware:
md5 = hashlib.md5(firmware.read()).hexdigest()
firmware.seek(0)
encoder = MultipartEncoder(fields={
'MD5': md5,
'file': (firmware_path, firmware, 'application/octet-stream')}
)
bar = tqdm(desc='Upload Progress',
total=encoder.len,
dynamic_ncols=True,
unit='B',
unit_scale=True,
unit_divisor=1024
)
monitor = MultipartEncoderMonitor(encoder, lambda monitor: bar.update(monitor.bytes_read - bar.n))
post_headers = {
'Host': host_ip,
'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/118.0',
'Accept': '*/*',
'Accept-Language': 'de,en-US;q=0.7,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Referer': f'{emsesp_url}',
'Connection': 'keep-alive',
'Content-Type': monitor.content_type,
'Content-Length': str(monitor.len),
'Origin': f'{emsesp_url}',
'Authorization': 'Bearer ' + f'{access_token}'
}
upload_url = f"{emsesp_url}/rest/uploadFile"
response = requests.post(upload_url, data=monitor, headers=post_headers, auth=None)
bar.close()
time.sleep(0.1)
print()
if response.status_code != 200:
print_fail("Upload failed (code " + response.status.code + ").")
else:
print_success("Upload successful.")
print()
env.Replace(UPLOADCMD=on_upload)

119
scripts/upload_cli.py Normal file
View File

@@ -0,0 +1,119 @@
# Modified from https://github.com/ayushsharma82/ElegantOTA
#
# Requires Python (sudo apt install python3-pip)
# `python3 -m venv venv` to create the virtual environment
# `source ./venv/bin/activate` to enter it
# `pip install -r requirements.txt` to install the libraries
#
# Run using for example:
# python3 upload_cli.py -i 10.10.10.173 -f ../build/firmware/EMS-ESP-3_7_0-dev_8-ESP32_4M.bin
import argparse
import requests
import hashlib
from urllib.parse import urlparse
import time
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
from tqdm import tqdm
from termcolor import cprint
def print_success(x): return cprint(x, 'green')
def print_fail(x): return cprint(x, 'red')
def upload(file, ip, username, password):
# Print welcome message
print()
print("EMS-ESP Firmware Upload")
# first check authentication
emsesp_url = "http://" + f'{ip}'
parsed_url = urlparse(emsesp_url)
host_ip = parsed_url.netloc
signon_url = f"{emsesp_url}/rest/signIn"
signon_headers = {
'Host': host_ip,
'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/118.0',
'Accept': '*/*',
'Accept-Language': 'de,en-US;q=0.7,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Referer': f'{emsesp_url}',
'Content-Type': 'application/json',
'Connection': 'keep-alive'
}
username_password = {
"username": username,
"password": password
}
response = requests.post(signon_url, json=username_password, headers=signon_headers, auth=None)
if response.status_code != 200:
print_fail("Authentication failed (code " + str(response.status_code) + ")")
return
print_success("Authentication successful")
access_token = response.json().get('access_token')
# start the upload
with open(file, 'rb') as firmware:
md5 = hashlib.md5(firmware.read()).hexdigest()
firmware.seek(0)
encoder = MultipartEncoder(fields={
'MD5': md5,
'file': (file, firmware, 'application/octet-stream')}
)
bar = tqdm(desc='Upload Progress',
total=encoder.len,
dynamic_ncols=True,
unit='B',
unit_scale=True,
unit_divisor=1024
)
monitor = MultipartEncoderMonitor(encoder, lambda monitor: bar.update(monitor.bytes_read - bar.n))
post_headers = {
'Host': host_ip,
'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/118.0',
'Accept': '*/*',
'Accept-Language': 'de,en-US;q=0.7,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Referer': f'{emsesp_url}',
'Connection': 'keep-alive',
'Content-Type': monitor.content_type,
'Content-Length': str(monitor.len),
'Origin': f'{emsesp_url}',
'Authorization': 'Bearer ' + f'{access_token}'
}
upload_url = f"{emsesp_url}/rest/uploadFile"
response = requests.post(upload_url, data=monitor, headers=post_headers, auth=None)
bar.close()
time.sleep(0.1)
if response.status_code != 200:
print_fail("Upload failed (code " + response.status.code + ").")
else:
print_success("Upload successful.")
print()
# main
parser = argparse.ArgumentParser(description="EMS-ESP Firmware Upload")
parser.add_argument("-f", "--file", metavar="FILE", required=True, type=str, help="firmware file")
parser.add_argument("-i", "--ip", metavar="IP", type=str, default="ems-esp.local", help="IP address of EMS-ESP")
parser.add_argument("-u", "--username", metavar="USERNAME", type=str, default="admin", help="admin user")
parser.add_argument("-p", "--password", metavar="PASSWORD", type=str, default="admin", help="admin password")
args = parser.parse_args()
upload(**vars(args))

View File

@@ -1,15 +0,0 @@
#!/usr/bin/env python
import subprocess
import os, argparse
print("\n** Starting upload...")
ap = argparse.ArgumentParser()
ap.add_argument("-p", "--port", required=True, help="port")
args = vars(ap.parse_args())
# esptool.py --chip esp32 --port "COM4" --baud 921600 --before default_reset --after hard_reset write_flash -z --flash_mode dio --flash_freq 40m --flash_size detect 0x1000 bootloader_dio_40m.bin 0x8000 partitions.bin 0xe000 boot_app0.bin 0x10000 EMS-ESP-dev-esp32.bin
subprocess.call(["esptool.py", "--chip esp32", "-p", args['port'], "--baud", "921600", "--before", "default_reset", "--after", "hard_reset", "write_flash", "-z", "--flash_mode", "dio", "--flash_freq", "40m","--flash_size", "detect", "0x1000", "bootloader_dio_40m.bin", "0x8000", "partitions.bin","0xe000", "boot_app0.bin", "0x10000", "EMS-ESP-dev-esp32.bin"])
print("\n** Finished upload.")