get.py

#!/usr/bin/env python

"""Script to download and extract tools

This script will download and extract required tools into the current directory.
Tools list is obtained from package/package_esp32_index.template.json file.
"""

from __future__ importprint_function

__author__="Ivan Grokhotkov"
__version__="2015"

importos
importshutil
importerrno
importos.path
importhashlib
importjson
importplatform
importsys
importtarfile
importzipfile
importre
importtime
importargparse

# Initialize start_time globally
start_time=-1

ifsys.version_info[0] ==3:
fromurllib.requestimporturlretrieve
fromurllib.requestimporturlopen

unicode=lambdas: str(s) # noqa: E731
else:
# Not Python 3 - today, it is most likely to be Python 2
fromurllibimporturlretrieve
fromurllibimporturlopen

if"Windows"inplatform.system():
importrequests

# determine if application is a script file or frozen exe
ifgetattr(sys, "frozen", False):
current_dir=os.path.dirname(os.path.realpath(unicode(sys.executable)))
elif__file__:
current_dir=os.path.dirname(os.path.realpath(unicode(__file__)))

dist_dir=current_dir+"/dist/"


defsha256sum(filename, blocksize=65536):
hash=hashlib.sha256()
withopen(filename, "rb") asf:
forblockiniter(lambda: f.read(blocksize), b""):
hash.update(block)
returnhash.hexdigest()


defmkdir_p(path):
try:
os.makedirs(path)
exceptOSErrorasexc:
ifexc.errno!=errno.EEXISTornotos.path.isdir(path):
raise


defformat_time(seconds):
minutes, seconds=divmod(seconds, 60)
return"{:02}:{:05.2f}".format(int(minutes), seconds)


defreport_progress(block_count, block_size, total_size, start_time):
downloaded_size=block_count*block_size
time_elapsed=time.time() -start_time
current_speed=downloaded_size/ (time_elapsed)

ifsys.stdout.isatty():
iftotal_size>0:
percent_complete=min((downloaded_size/total_size) *100, 100)
sys.stdout.write(
f"\rDownloading... {percent_complete:.2f}% - {downloaded_size/1024/1024:.2f} MB downloaded - Elapsed Time: {format_time(time_elapsed)} - Speed: {current_speed/1024/1024:.2f} MB/s"# noqa: E501
 )
else:
sys.stdout.write(
f"\rDownloading... {downloaded_size/1024/1024:.2f} MB downloaded - Elapsed Time: {format_time(time_elapsed)} - Speed: {current_speed/1024/1024:.2f} MB/s"# noqa: E501
 )
sys.stdout.flush()


defprint_verification_progress(total_files, i, t1):
ifsys.stdout.isatty():
sys.stdout.write(f"\rElapsed time {format_time(time.time() -t1)}")
sys.stdout.flush()


defverify_files(filename, destination, rename_to):
# Set the path of the extracted directory
extracted_dir_path=destination
t1=time.time()
iffilename.endswith(".zip"):
try:
archive=zipfile.ZipFile(filename, "r")
file_list=archive.namelist()
exceptzipfile.BadZipFile:
ifverbose:
print(f"Verification failed; aborted in {format_time(time.time() -t1)}")
returnFalse
eliffilename.endswith(".tar.gz"):
try:
archive=tarfile.open(filename, "r:gz")
file_list=archive.getnames()
excepttarfile.ReadError:
ifverbose:
print(f"Verification failed; aborted in {format_time(time.time() -t1)}")
returnFalse
eliffilename.endswith(".tar.xz"):
try:
archive=tarfile.open(filename, "r:xz")
file_list=archive.getnames()
excepttarfile.ReadError:
ifverbose:
print(f"Verification failed; aborted in {format_time(time.time() -t1)}")
returnFalse
else:
raiseNotImplementedError("Unsupported archive type")

try:
first_dir=file_list[0].split("/")[0]
total_files=len(file_list)
fori, zipped_fileinenumerate(file_list, 1):
local_path=os.path.join(extracted_dir_path, zipped_file.replace(first_dir, rename_to, 1))
ifnotos.path.exists(local_path):
ifverbose:
print(f"\nMissing {zipped_file} on location: {extracted_dir_path}")
print(f"Verification failed; aborted in {format_time(time.time() -t1)}")
returnFalse
print_verification_progress(total_files, i, t1)
exceptExceptionase:
print(f"\nError: {e}")
returnFalse

ifverbose:
print(f"\nVerification passed; completed in {format_time(time.time() -t1)}")

returnTrue


defis_latest_version(destination, dirname, rename_to, cfile, checksum):
current_version=None
expected_version=None

try:
expected_version=checksum
withopen(os.path.join(destination, rename_to, ".package_checksum"), "r") asf:
current_version=f.read()

ifverbose:
print(f"\nTool: {rename_to}")
print(f"Current version: {current_version}")
print(f"Expected version: {expected_version}")

ifcurrent_versionandcurrent_version==expected_version:
ifverbose:
print("Latest version already installed. Skipping extraction")
returnTrue

ifverbose:
print("New version detected")

exceptExceptionase:
ifverbose:
print(f"Failed to verify version for {rename_to}: {e}")

returnFalse


defunpack(filename, destination, force_extract, checksum): # noqa: C901
sys_name=platform.system()
dirname=""
cfile=None# Compressed file
file_is_corrupted=False
ifnotforce_extract:
print(" > Verify archive... ", end="", flush=True)

try:
iffilename.endswith("tar.gz"):
iftarfile.is_tarfile(filename):
cfile=tarfile.open(filename, "r:gz")
dirname=cfile.getnames()[0].split("/")[0]
else:
print("File corrupted!")
file_is_corrupted=True
eliffilename.endswith("tar.xz"):
iftarfile.is_tarfile(filename):
cfile=tarfile.open(filename, "r:xz")
dirname=cfile.getnames()[0].split("/")[0]
else:
print("File corrupted!")
file_is_corrupted=True
eliffilename.endswith("zip"):
ifzipfile.is_zipfile(filename):
cfile=zipfile.ZipFile(filename)
dirname=cfile.namelist()[0].split("/")[0]
else:
print("File corrupted!")
file_is_corrupted=True
else:
raiseNotImplementedError("Unsupported archive type")
exceptEOFError:
print("File corrupted or incomplete!")
cfile=None
file_is_corrupted=True

iffile_is_corrupted:
corrupted_filename=filename+".corrupted"
os.rename(filename, corrupted_filename)
ifverbose:
print(f"Renaming corrupted archive to {corrupted_filename}")
returnFalse

# A little trick to rename tool directories so they don't contain version number
rename_to=re.match(r"^([a-z][^\-]*\-*)+", dirname).group(0).strip("-")
ifrename_to==dirnameanddirname.startswith("esp32-arduino-libs-"):
rename_to="esp32-arduino-libs"
elifrename_to==dirnameanddirname.startswith("esptool-"):
rename_to="esptool"

ifnotforce_extract:
ifis_latest_version(destination, dirname, rename_to, cfile, checksum):
ifverify_files(filename, destination, rename_to):
print(" Files ok. Skipping Extraction")
returnTrue
print(" Extracting archive...")
else:
print(" Forcing extraction")

ifos.path.isdir(os.path.join(destination, rename_to)):
print("Removing existing {0} ...".format(rename_to))
shutil.rmtree(os.path.join(destination, rename_to), ignore_errors=True)

iffilename.endswith("tar.gz"):
ifnotcfile:
cfile=tarfile.open(filename, "r:gz")
cfile.extractall(destination, filter="tar")
eliffilename.endswith("tar.xz"):
ifnotcfile:
cfile=tarfile.open(filename, "r:xz")
cfile.extractall(destination, filter="tar")
eliffilename.endswith("zip"):
ifnotcfile:
cfile=zipfile.ZipFile(filename)
cfile.extractall(destination)
else:
raiseNotImplementedError("Unsupported archive type")

ifrename_to!=dirname:
print("Renaming {0} to {1} ...".format(dirname, rename_to))
shutil.move(dirname, rename_to)

# Add execute permission to esptool on non-Windows platforms
ifrename_to.startswith("esptool") and"CYGWIN_NT"notinsys_nameand"Windows"notinsys_name:
st=os.stat(os.path.join(destination, rename_to, "esptool"))
os.chmod(os.path.join(destination, rename_to, "esptool"), st.st_mode|0o111)

withopen(os.path.join(destination, rename_to, ".package_checksum"), "w") asf:
f.write(checksum)

ifverify_files(filename, destination, rename_to):
print(" Files extracted successfully.")
returnTrue
else:
print(" Failed to extract files.")
returnFalse


defdownload_file_with_progress(url, filename, start_time):
importssl
importcontextlib

ctx=ssl.create_default_context()
ctx.check_hostname=False
ctx.verify_mode=ssl.CERT_NONE
withcontextlib.closing(urlopen(url, context=ctx)) asfp:
total_size=int(fp.getheader("Content-Length", fp.getheader("Content-length", "0")))
block_count=0
block_size=1024*8
block=fp.read(block_size)
ifblock:
withopen(filename, "wb") asout_file:
out_file.write(block)
block_count+=1
report_progress(block_count, block_size, total_size, start_time)
whileTrue:
block=fp.read(block_size)
ifnotblock:
break
out_file.write(block)
block_count+=1
report_progress(block_count, block_size, total_size, start_time)
else:
raiseException("Non-existing file or connection error")


defdownload_file(url, filename):
importssl
importcontextlib

ctx=ssl.create_default_context()
ctx.check_hostname=False
ctx.verify_mode=ssl.CERT_NONE
withcontextlib.closing(urlopen(url, context=ctx)) asfp:
block_size=1024*8
block=fp.read(block_size)
ifblock:
withopen(filename, "wb") asout_file:
out_file.write(block)
whileTrue:
block=fp.read(block_size)
ifnotblock:
break
out_file.write(block)
else:
raiseException("Non-existing file or connection error")


defget_tool(tool, force_download, force_extract):
sys_name=platform.system()
archive_name=tool["archiveFileName"]
checksum=tool["checksum"][8:]
local_path=dist_dir+archive_name
url=tool["url"]
start_time=time.time()
print("")
ifnotos.path.isfile(local_path) orforce_download:
ifverbose:
print("Downloading '"+archive_name+"' to '"+local_path+"'")
else:
print("Downloading '"+archive_name+"' ...")
sys.stdout.flush()
if"CYGWIN_NT"insys_name:
importssl

ctx=ssl.create_default_context()
ctx.check_hostname=False
ctx.verify_mode=ssl.CERT_NONE
urlretrieve(url, local_path, report_progress, context=ctx)
elif"Windows"insys_name:
r=requests.get(url)
f=open(local_path, "wb")
f.write(r.content)
f.close()
else:
is_ci=os.environ.get("GITHUB_WORKSPACE")
ifis_ci:
download_file(url, local_path)
else:
try:
urlretrieve(url, local_path, report_progress)
except: # noqa: E722
download_file_with_progress(url, local_path, start_time)
sys.stdout.write(" - Done\n")
sys.stdout.flush()
else:
print("Tool {0} already downloaded".format(archive_name))
sys.stdout.flush()

ifsha256sum(local_path) !=checksum:
print("Checksum mismatch for {0}".format(archive_name))
returnFalse

returnunpack(local_path, ".", force_extract, checksum)


defload_tools_list(filename, platform):
tools_info=json.load(open(filename))["packages"][0]["tools"]
tools_to_download= []
fortintools_info:
ifplatform=="x86_64-mingw32":
if"i686-mingw32"notin [p["host"] forpint["systems"]]:
raiseException("Windows x64 requires both i686-mingw32 and x86_64-mingw32 tools")

tool_platform= [pforpint["systems"] ifp["host"] ==platform]
iflen(tool_platform) ==0:
# Fallback to x86 on Apple ARM
ifplatform=="arm64-apple-darwin":
tool_platform= [pforpint["systems"] ifp["host"] =="x86_64-apple-darwin"]
iflen(tool_platform) ==0:
continue
# Fallback to 32bit on 64bit x86 Windows
elifplatform=="x86_64-mingw32":
tool_platform= [pforpint["systems"] ifp["host"] =="i686-mingw32"]
iflen(tool_platform) ==0:
continue
else:
ifverbose:
print(f"Tool {t['name']} is not available for platform {platform}")
continue
tools_to_download.append(tool_platform[0])
returntools_to_download


defidentify_platform():
arduino_platform_names= {
"Darwin": {32: "i386-apple-darwin", 64: "x86_64-apple-darwin"},
"DarwinARM": {32: "arm64-apple-darwin", 64: "arm64-apple-darwin"},
"Linux": {32: "i686-pc-linux-gnu", 64: "x86_64-pc-linux-gnu"},
"LinuxARM": {32: "arm-linux-gnueabihf", 64: "aarch64-linux-gnu"},
"Windows": {32: "i686-mingw32", 64: "x86_64-mingw32"},
 }
bits=32
ifsys.maxsize>2**32:
bits=64
sys_name=platform.system()
sys_platform=platform.platform()
if"Darwin"insys_nameand (sys_platform.find("arm") >0orsys_platform.find("arm64") >0):
sys_name="DarwinARM"
if"Linux"insys_nameand (sys_platform.find("arm") >0orsys_platform.find("aarch64") >0):
sys_name="LinuxARM"
if"CYGWIN_NT"insys_name:
sys_name="Windows"
print("System: %s, Bits: %d, Info: %s"% (sys_name, bits, sys_platform))
returnarduino_platform_names[sys_name][bits]


if__name__=="__main__":
parser=argparse.ArgumentParser(description="Download and extract tools")

parser.add_argument("-v", "--verbose", action="store_true", required=False, help="Print verbose output")

parser.add_argument("-d", "--force_download", action="store_true", required=False, help="Force download of tools")

parser.add_argument("-e", "--force_extract", action="store_true", required=False, help="Force extraction of tools")

parser.add_argument(
"-f", "--force_all", action="store_true", required=False, help="Force download and extraction of tools"
 )

parser.add_argument("-t", "--test", action="store_true", required=False, help=argparse.SUPPRESS)

args=parser.parse_args()

verbose=args.verbose
force_download=args.force_download
force_extract=args.force_extract
force_all=args.force_all
is_test=args.test

# Set current directory to the script location
ifgetattr(sys, "frozen", False):
os.chdir(os.path.dirname(sys.executable))
else:
os.chdir(os.path.dirname(os.path.abspath(__file__)))

ifis_testand (force_downloadorforce_extractorforce_all):
print("Cannot combine test (-t) and forced execution (-d | -e | -f)")
parser.print_help(sys.stderr)
sys.exit(1)

ifis_test:
print("Test run!")

ifforce_all:
force_download=True
force_extract=True

identified_platform=identify_platform()
print("Platform: {0}".format(identified_platform))
tools_to_download=load_tools_list(
current_dir+"/../package/package_esp32_index.template.json", identified_platform
 )
mkdir_p(dist_dir)

print("\nDownloading and extracting tools...")

fortoolintools_to_download:
ifis_test:
print("Would install: {0}".format(tool["archiveFileName"]))
else:
ifnotget_tool(tool, force_download, force_extract):
ifverbose:
print(f"Tool {tool['archiveFileName']} was corrupted. Re-downloading...\n")
ifnotget_tool(tool, True, force_extract):
print(f"Tool {tool['archiveFileName']} was corrupted, but re-downloading did not help!\n")
sys.exit(1)

print("\nPlatform Tools Installed")