platformio_upload.py

# From https://github.com/ayushsharma82/AsyncElegantOTA/blob/master/platformio_upload.py
# Allows PlatformIO to upload directly to AsyncElegantOTA
#
# To use:
# - copy this script into the same folder as your platformio.ini
# - set the following for your project in platformio.ini:
#
# extra_scripts = platformio_upload.py
# upload_protocol = custom
# upload_url = <your upload URL>
# 
# An example of an upload URL:
# upload_URL = http://192.168.1.123/update

import requests
import hashlib
Import('env')

try:
    from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
    from tqdm import tqdm
except ImportError:
    env.Execute("$PYTHONEXE -m pip install requests_toolbelt")
    env.Execute("$PYTHONEXE -m pip install tqdm")
    from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor
    from tqdm import tqdm

def on_upload(source, target, env):
    firmware_path = str(source[0])
    upload_url = env.GetProjectOption('upload_url')

    with open(firmware_path, 'rb') as firmware:
        md5 = hashlib.md5(firmware.read()).hexdigest()
        firmware.seek(0)
        encoder = MultipartEncoder(fields={
            'MD5': md5, 
            'firmware': ('firmware', 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))

        try:
            response = requests.post(upload_url, data=monitor, headers={'Content-Type': monitor.content_type}, timeout=180)
            bar.close()
        
            # Basic error checking
            if response.status_code != 200 or 'error' in response.text.lower(): 
                raise Exception(f"Upload failed with status {response.status_code}: {response.text}")
            else:
                print("Upload completed successfully!")
                print(response.text)
        except requests.Timeout:
            print("Request timed out!")
        except requests.RequestException as e:
            print(f"Error occurred: {e}")


            
env.Replace(UPLOADCMD=on_upload)