aboutsummaryrefslogtreecommitdiff
path: root/tools
diff options
context:
space:
mode:
authorDamien George2020-02-27 15:36:53 +1100
committerDamien George2020-02-28 10:33:03 +1100
commit69661f3343bedf86e514337cff63d96cc42f8859 (patch)
treeaf5dfb380ffdb75dda84828f63cf9d840d992f0f /tools
parent3f39d18c2b884d32f0443e2e8114ff9d7a14d718 (diff)
all: Reformat C and Python source code with tools/codeformat.py.
This is run with uncrustify 0.70.1, and black 19.10b0.
Diffstat (limited to 'tools')
-rwxr-xr-xtools/dfu.py260
-rw-r--r--tools/file2h.py12
-rw-r--r--tools/gen-cpydiff.py192
-rw-r--r--tools/gendoc.py245
-rw-r--r--tools/insert-usb-ids.py16
-rwxr-xr-xtools/make-frozen.py11
-rw-r--r--tools/makemanifest.py122
-rwxr-xr-xtools/metrics.py87
-rwxr-xr-xtools/mpy-tool.py635
-rwxr-xr-xtools/mpy_cross_all.py16
-rwxr-xr-xtools/mpy_ld.py683
-rwxr-xr-xtools/pyboard.py269
-rwxr-xr-xtools/pydfu.py300
-rwxr-xr-xtools/tinytest-codegen.py106
-rwxr-xr-xtools/uf2conv.py170
-rw-r--r--tools/upip.py64
-rw-r--r--tools/upip_utarfile.py45
17 files changed, 1890 insertions, 1343 deletions
diff --git a/tools/dfu.py b/tools/dfu.py
index ee89c7541..6591436e9 100755
--- a/tools/dfu.py
+++ b/tools/dfu.py
@@ -4,122 +4,160 @@
# Distributed under Gnu LGPL 3.0
# see http://www.gnu.org/licenses/lgpl-3.0.txt
-import sys,struct,zlib,os
+import sys, struct, zlib, os
from optparse import OptionParser
-DEFAULT_DEVICE="0x0483:0xdf11"
+DEFAULT_DEVICE = "0x0483:0xdf11"
+
+
+def named(tuple, names):
+ return dict(zip(names.split(), tuple))
+
+
+def consume(fmt, data, names):
+ n = struct.calcsize(fmt)
+ return named(struct.unpack(fmt, data[:n]), names), data[n:]
+
-def named(tuple,names):
- return dict(zip(names.split(),tuple))
-def consume(fmt,data,names):
- n = struct.calcsize(fmt)
- return named(struct.unpack(fmt,data[:n]),names),data[n:]
def cstring(string):
- return string.split('\0',1)[0]
+ return string.split("\0", 1)[0]
+
+
def compute_crc(data):
- return 0xFFFFFFFF & -zlib.crc32(data) -1
-
-def parse(file,dump_images=False):
- print ('File: "%s"' % file)
- data = open(file,'rb').read()
- crc = compute_crc(data[:-4])
- prefix, data = consume('<5sBIB',data,'signature version size targets')
- print ('%(signature)s v%(version)d, image size: %(size)d, targets: %(targets)d' % prefix)
- for t in range(prefix['targets']):
- tprefix, data = consume('<6sBI255s2I',data,'signature altsetting named name size elements')
- tprefix['num'] = t
- if tprefix['named']:
- tprefix['name'] = cstring(tprefix['name'])
- else:
- tprefix['name'] = ''
- print ('%(signature)s %(num)d, alt setting: %(altsetting)s, name: "%(name)s", size: %(size)d, elements: %(elements)d' % tprefix)
- tsize = tprefix['size']
- target, data = data[:tsize], data[tsize:]
- for e in range(tprefix['elements']):
- eprefix, target = consume('<2I',target,'address size')
- eprefix['num'] = e
- print (' %(num)d, address: 0x%(address)08x, size: %(size)d' % eprefix)
- esize = eprefix['size']
- image, target = target[:esize], target[esize:]
- if dump_images:
- out = '%s.target%d.image%d.bin' % (file,t,e)
- open(out,'wb').write(image)
- print (' DUMPED IMAGE TO "%s"' % out)
- if len(target):
- print ("target %d: PARSE ERROR" % t)
- suffix = named(struct.unpack('<4H3sBI',data[:16]),'device product vendor dfu ufd len crc')
- print ('usb: %(vendor)04x:%(product)04x, device: 0x%(device)04x, dfu: 0x%(dfu)04x, %(ufd)s, %(len)d, 0x%(crc)08x' % suffix)
- if crc != suffix['crc']:
- print ("CRC ERROR: computed crc32 is 0x%08x" % crc)
- data = data[16:]
- if data:
- print ("PARSE ERROR")
-
-def build(file,targets,device=DEFAULT_DEVICE):
- data = b''
- for t,target in enumerate(targets):
- tdata = b''
- for image in target:
- # pad image to 8 bytes (needed at least for L476)
- pad = (8 - len(image['data']) % 8 ) % 8
- image['data'] = image['data'] + bytes(bytearray(8)[0:pad])
- #
- tdata += struct.pack('<2I',image['address'],len(image['data']))+image['data']
- tdata = struct.pack('<6sBI255s2I',b'Target',0,1, b'ST...',len(tdata),len(target)) + tdata
- data += tdata
- data = struct.pack('<5sBIB',b'DfuSe',1,len(data)+11,len(targets)) + data
- v,d=map(lambda x: int(x,0) & 0xFFFF, device.split(':',1))
- data += struct.pack('<4H3sB',0,d,v,0x011a,b'UFD',16)
- crc = compute_crc(data)
- data += struct.pack('<I',crc)
- open(file,'wb').write(data)
-
-if __name__=="__main__":
- usage = """
+ return 0xFFFFFFFF & -zlib.crc32(data) - 1
+
+
+def parse(file, dump_images=False):
+ print('File: "%s"' % file)
+ data = open(file, "rb").read()
+ crc = compute_crc(data[:-4])
+ prefix, data = consume("<5sBIB", data, "signature version size targets")
+ print("%(signature)s v%(version)d, image size: %(size)d, targets: %(targets)d" % prefix)
+ for t in range(prefix["targets"]):
+ tprefix, data = consume(
+ "<6sBI255s2I", data, "signature altsetting named name size elements"
+ )
+ tprefix["num"] = t
+ if tprefix["named"]:
+ tprefix["name"] = cstring(tprefix["name"])
+ else:
+ tprefix["name"] = ""
+ print(
+ '%(signature)s %(num)d, alt setting: %(altsetting)s, name: "%(name)s", size: %(size)d, elements: %(elements)d'
+ % tprefix
+ )
+ tsize = tprefix["size"]
+ target, data = data[:tsize], data[tsize:]
+ for e in range(tprefix["elements"]):
+ eprefix, target = consume("<2I", target, "address size")
+ eprefix["num"] = e
+ print(" %(num)d, address: 0x%(address)08x, size: %(size)d" % eprefix)
+ esize = eprefix["size"]
+ image, target = target[:esize], target[esize:]
+ if dump_images:
+ out = "%s.target%d.image%d.bin" % (file, t, e)
+ open(out, "wb").write(image)
+ print(' DUMPED IMAGE TO "%s"' % out)
+ if len(target):
+ print("target %d: PARSE ERROR" % t)
+ suffix = named(struct.unpack("<4H3sBI", data[:16]), "device product vendor dfu ufd len crc")
+ print(
+ "usb: %(vendor)04x:%(product)04x, device: 0x%(device)04x, dfu: 0x%(dfu)04x, %(ufd)s, %(len)d, 0x%(crc)08x"
+ % suffix
+ )
+ if crc != suffix["crc"]:
+ print("CRC ERROR: computed crc32 is 0x%08x" % crc)
+ data = data[16:]
+ if data:
+ print("PARSE ERROR")
+
+
+def build(file, targets, device=DEFAULT_DEVICE):
+ data = b""
+ for t, target in enumerate(targets):
+ tdata = b""
+ for image in target:
+ # pad image to 8 bytes (needed at least for L476)
+ pad = (8 - len(image["data"]) % 8) % 8
+ image["data"] = image["data"] + bytes(bytearray(8)[0:pad])
+ #
+ tdata += struct.pack("<2I", image["address"], len(image["data"])) + image["data"]
+ tdata = (
+ struct.pack("<6sBI255s2I", b"Target", 0, 1, b"ST...", len(tdata), len(target)) + tdata
+ )
+ data += tdata
+ data = struct.pack("<5sBIB", b"DfuSe", 1, len(data) + 11, len(targets)) + data
+ v, d = map(lambda x: int(x, 0) & 0xFFFF, device.split(":", 1))
+ data += struct.pack("<4H3sB", 0, d, v, 0x011A, b"UFD", 16)
+ crc = compute_crc(data)
+ data += struct.pack("<I", crc)
+ open(file, "wb").write(data)
+
+
+if __name__ == "__main__":
+ usage = """
%prog [-d|--dump] infile.dfu
%prog {-b|--build} address:file.bin [-b address:file.bin ...] [{-D|--device}=vendor:device] outfile.dfu"""
- parser = OptionParser(usage=usage)
- parser.add_option("-b", "--build", action="append", dest="binfiles",
- help="build a DFU file from given BINFILES", metavar="BINFILES")
- parser.add_option("-D", "--device", action="store", dest="device",
- help="build for DEVICE, defaults to %s" % DEFAULT_DEVICE, metavar="DEVICE")
- parser.add_option("-d", "--dump", action="store_true", dest="dump_images",
- default=False, help="dump contained images to current directory")
- (options, args) = parser.parse_args()
-
- if options.binfiles and len(args)==1:
- target = []
- for arg in options.binfiles:
- try:
- address,binfile = arg.split(':',1)
- except ValueError:
- print ("Address:file couple '%s' invalid." % arg)
- sys.exit(1)
- try:
- address = int(address,0) & 0xFFFFFFFF
- except ValueError:
- print ("Address %s invalid." % address)
- sys.exit(1)
- if not os.path.isfile(binfile):
- print ("Unreadable file '%s'." % binfile)
+ parser = OptionParser(usage=usage)
+ parser.add_option(
+ "-b",
+ "--build",
+ action="append",
+ dest="binfiles",
+ help="build a DFU file from given BINFILES",
+ metavar="BINFILES",
+ )
+ parser.add_option(
+ "-D",
+ "--device",
+ action="store",
+ dest="device",
+ help="build for DEVICE, defaults to %s" % DEFAULT_DEVICE,
+ metavar="DEVICE",
+ )
+ parser.add_option(
+ "-d",
+ "--dump",
+ action="store_true",
+ dest="dump_images",
+ default=False,
+ help="dump contained images to current directory",
+ )
+ (options, args) = parser.parse_args()
+
+ if options.binfiles and len(args) == 1:
+ target = []
+ for arg in options.binfiles:
+ try:
+ address, binfile = arg.split(":", 1)
+ except ValueError:
+ print("Address:file couple '%s' invalid." % arg)
+ sys.exit(1)
+ try:
+ address = int(address, 0) & 0xFFFFFFFF
+ except ValueError:
+ print("Address %s invalid." % address)
+ sys.exit(1)
+ if not os.path.isfile(binfile):
+ print("Unreadable file '%s'." % binfile)
+ sys.exit(1)
+ target.append({"address": address, "data": open(binfile, "rb").read()})
+ outfile = args[0]
+ device = DEFAULT_DEVICE
+ if options.device:
+ device = options.device
+ try:
+ v, d = map(lambda x: int(x, 0) & 0xFFFF, device.split(":", 1))
+ except:
+ print("Invalid device '%s'." % device)
+ sys.exit(1)
+ build(outfile, [target], device)
+ elif len(args) == 1:
+ infile = args[0]
+ if not os.path.isfile(infile):
+ print("Unreadable file '%s'." % infile)
+ sys.exit(1)
+ parse(infile, dump_images=options.dump_images)
+ else:
+ parser.print_help()
sys.exit(1)
- target.append({ 'address': address, 'data': open(binfile,'rb').read() })
- outfile = args[0]
- device = DEFAULT_DEVICE
- if options.device:
- device=options.device
- try:
- v,d=map(lambda x: int(x,0) & 0xFFFF, device.split(':',1))
- except:
- print ("Invalid device '%s'." % device)
- sys.exit(1)
- build(outfile,[target],device)
- elif len(args)==1:
- infile = args[0]
- if not os.path.isfile(infile):
- print ("Unreadable file '%s'." % infile)
- sys.exit(1)
- parse(infile, dump_images=options.dump_images)
- else:
- parser.print_help()
- sys.exit(1)
diff --git a/tools/file2h.py b/tools/file2h.py
index 2a04ae22b..4607ab927 100644
--- a/tools/file2h.py
+++ b/tools/file2h.py
@@ -12,19 +12,19 @@ import sys
# Can either be set explicitly, or left blank to auto-detect
# Except auto-detect doesn't work because the file has been passed
# through Python text processing, which makes all EOL a \n
-line_end = '\\r\\n'
+line_end = "\\r\\n"
if __name__ == "__main__":
filename = sys.argv[1]
- for line in open(filename, 'r').readlines():
+ for line in open(filename, "r").readlines():
if not line_end:
- for ending in ('\r\n', '\r', '\n'):
+ for ending in ("\r\n", "\r", "\n"):
if line.endswith(ending):
- line_end = ending.replace('\r', '\\r').replace('\n', '\\n')
+ line_end = ending.replace("\r", "\\r").replace("\n", "\\n")
break
if not line_end:
raise Exception("Couldn't auto-detect line-ending of %s" % filename)
- line = line.rstrip('\r\n')
- line = line.replace('\\', '\\\\')
+ line = line.rstrip("\r\n")
+ line = line.replace("\\", "\\\\")
line = line.replace('"', '\\"')
print('"%s%s"' % (line, line_end))
diff --git a/tools/gen-cpydiff.py b/tools/gen-cpydiff.py
index 356ade89d..c45bad105 100644
--- a/tools/gen-cpydiff.py
+++ b/tools/gen-cpydiff.py
@@ -37,89 +37,128 @@ from collections import namedtuple
# such version should be used to test for differences. If your default python3
# executable is of lower version, you can point MICROPY_CPYTHON3 environment var
# to the correct executable.
-if os.name == 'nt':
- CPYTHON3 = os.getenv('MICROPY_CPYTHON3', 'python3.exe')
- MICROPYTHON = os.getenv('MICROPY_MICROPYTHON', '../ports/windows/micropython.exe')
+if os.name == "nt":
+ CPYTHON3 = os.getenv("MICROPY_CPYTHON3", "python3.exe")
+ MICROPYTHON = os.getenv("MICROPY_MICROPYTHON", "../ports/windows/micropython.exe")
else:
- CPYTHON3 = os.getenv('MICROPY_CPYTHON3', 'python3')
- MICROPYTHON = os.getenv('MICROPY_MICROPYTHON', '../ports/unix/micropython')
-
-TESTPATH = '../tests/cpydiff/'
-DOCPATH = '../docs/genrst/'
-INDEXTEMPLATE = '../docs/differences/index_template.txt'
-INDEX = 'index.rst'
-
-HEADER = '.. This document was generated by tools/gen-cpydiff.py\n\n'
-UIMPORTLIST = {'struct', 'collections', 'json'}
-CLASSMAP = {'Core': 'Core language', 'Types': 'Builtin types'}
-INDEXPRIORITY = ['syntax', 'core_language', 'builtin_types', 'modules']
-RSTCHARS = ['=', '-', '~', '`', ':']
+ CPYTHON3 = os.getenv("MICROPY_CPYTHON3", "python3")
+ MICROPYTHON = os.getenv("MICROPY_MICROPYTHON", "../ports/unix/micropython")
+
+TESTPATH = "../tests/cpydiff/"
+DOCPATH = "../docs/genrst/"
+INDEXTEMPLATE = "../docs/differences/index_template.txt"
+INDEX = "index.rst"
+
+HEADER = ".. This document was generated by tools/gen-cpydiff.py\n\n"
+UIMPORTLIST = {"struct", "collections", "json"}
+CLASSMAP = {"Core": "Core language", "Types": "Builtin types"}
+INDEXPRIORITY = ["syntax", "core_language", "builtin_types", "modules"]
+RSTCHARS = ["=", "-", "~", "`", ":"]
SPLIT = '"""\n|categories: |description: |cause: |workaround: '
-TAB = ' '
+TAB = " "
+
+Output = namedtuple(
+ "output",
+ [
+ "name",
+ "class_",
+ "desc",
+ "cause",
+ "workaround",
+ "code",
+ "output_cpy",
+ "output_upy",
+ "status",
+ ],
+)
-Output = namedtuple('output', ['name', 'class_', 'desc', 'cause', 'workaround', 'code',
- 'output_cpy', 'output_upy', 'status'])
def readfiles():
""" Reads test files """
- tests = list(filter(lambda x: x.endswith('.py'), os.listdir(TESTPATH)))
+ tests = list(filter(lambda x: x.endswith(".py"), os.listdir(TESTPATH)))
tests.sort()
files = []
for test in tests:
- text = open(TESTPATH + test, 'r').read()
+ text = open(TESTPATH + test, "r").read()
try:
- class_, desc, cause, workaround, code = [x.rstrip() for x in \
- list(filter(None, re.split(SPLIT, text)))]
- output = Output(test, class_, desc, cause, workaround, code, '', '', '')
+ class_, desc, cause, workaround, code = [
+ x.rstrip() for x in list(filter(None, re.split(SPLIT, text)))
+ ]
+ output = Output(test, class_, desc, cause, workaround, code, "", "", "")
files.append(output)
except IndexError:
- print('Incorrect format in file ' + TESTPATH + test)
+ print("Incorrect format in file " + TESTPATH + test)
return files
+
def uimports(code):
""" converts CPython module names into MicroPython equivalents """
for uimport in UIMPORTLIST:
- uimport = bytes(uimport, 'utf8')
- code = code.replace(uimport, b'u' + uimport)
+ uimport = bytes(uimport, "utf8")
+ code = code.replace(uimport, b"u" + uimport)
return code
+
def run_tests(tests):
""" executes all tests """
results = []
for test in tests:
- with open(TESTPATH + test.name, 'rb') as f:
+ with open(TESTPATH + test.name, "rb") as f:
input_cpy = f.read()
input_upy = uimports(input_cpy)
- process = subprocess.Popen(CPYTHON3, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
- output_cpy = [com.decode('utf8') for com in process.communicate(input_cpy)]
+ process = subprocess.Popen(
+ CPYTHON3,
+ shell=True,
+ stdout=subprocess.PIPE,
+ stdin=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ output_cpy = [com.decode("utf8") for com in process.communicate(input_cpy)]
- process = subprocess.Popen(MICROPYTHON, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
- output_upy = [com.decode('utf8') for com in process.communicate(input_upy)]
+ process = subprocess.Popen(
+ MICROPYTHON,
+ shell=True,
+ stdout=subprocess.PIPE,
+ stdin=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
+ output_upy = [com.decode("utf8") for com in process.communicate(input_upy)]
if output_cpy[0] == output_upy[0] and output_cpy[1] == output_upy[1]:
- status = 'Supported'
- print('Supported operation!\nFile: ' + TESTPATH + test.name)
+ status = "Supported"
+ print("Supported operation!\nFile: " + TESTPATH + test.name)
else:
- status = 'Unsupported'
+ status = "Unsupported"
- output = Output(test.name, test.class_, test.desc, test.cause,
- test.workaround, test.code, output_cpy, output_upy, status)
+ output = Output(
+ test.name,
+ test.class_,
+ test.desc,
+ test.cause,
+ test.workaround,
+ test.code,
+ output_cpy,
+ output_upy,
+ status,
+ )
results.append(output)
results.sort(key=lambda x: x.class_)
return results
+
def indent(block, spaces):
""" indents paragraphs of text for rst formatting """
- new_block = ''
- for line in block.split('\n'):
- new_block += spaces + line + '\n'
+ new_block = ""
+ for line in block.split("\n"):
+ new_block += spaces + line + "\n"
return new_block
+
def gen_table(contents):
""" creates a table given any set of columns """
xlengths = []
@@ -127,31 +166,32 @@ def gen_table(contents):
for column in contents:
col_len = 0
for entry in column:
- lines = entry.split('\n')
+ lines = entry.split("\n")
for line in lines:
col_len = max(len(line) + 2, col_len)
xlengths.append(col_len)
for i in range(len(contents[0])):
ymax = 0
for j in range(len(contents)):
- ymax = max(ymax, len(contents[j][i].split('\n')))
+ ymax = max(ymax, len(contents[j][i].split("\n")))
ylengths.append(ymax)
- table_divider = '+' + ''.join(['-' * i + '+' for i in xlengths]) + '\n'
+ table_divider = "+" + "".join(["-" * i + "+" for i in xlengths]) + "\n"
table = table_divider
for i in range(len(ylengths)):
row = [column[i] for column in contents]
- row = [entry + '\n' * (ylengths[i]-len(entry.split('\n'))) for entry in row]
- row = [entry.split('\n') for entry in row]
+ row = [entry + "\n" * (ylengths[i] - len(entry.split("\n"))) for entry in row]
+ row = [entry.split("\n") for entry in row]
for j in range(ylengths[i]):
k = 0
for entry in row:
width = xlengths[k]
- table += ''.join(['| {:{}}'.format(entry[j], width - 1)])
+ table += "".join(["| {:{}}".format(entry[j], width - 1)])
k += 1
- table += '|\n'
+ table += "|\n"
table += table_divider
- return table + '\n'
+ return table + "\n"
+
def gen_rst(results):
""" creates restructured text documents to display tests """
@@ -166,61 +206,63 @@ def gen_rst(results):
toctree = []
class_ = []
for output in results:
- section = output.class_.split(',')
+ section = output.class_.split(",")
for i in range(len(section)):
section[i] = section[i].rstrip()
if section[i] in CLASSMAP:
section[i] = CLASSMAP[section[i]]
if i >= len(class_) or section[i] != class_[i]:
if i == 0:
- filename = section[i].replace(' ', '_').lower()
- rst = open(DOCPATH + filename + '.rst', 'w')
+ filename = section[i].replace(" ", "_").lower()
+ rst = open(DOCPATH + filename + ".rst", "w")
rst.write(HEADER)
- rst.write(section[i] + '\n')
+ rst.write(section[i] + "\n")
rst.write(RSTCHARS[0] * len(section[i]))
rst.write(time.strftime("\nGenerated %a %d %b %Y %X UTC\n\n", time.gmtime()))
toctree.append(filename)
else:
- rst.write(section[i] + '\n')
- rst.write(RSTCHARS[min(i, len(RSTCHARS)-1)] * len(section[i]))
- rst.write('\n\n')
+ rst.write(section[i] + "\n")
+ rst.write(RSTCHARS[min(i, len(RSTCHARS) - 1)] * len(section[i]))
+ rst.write("\n\n")
class_ = section
- rst.write('.. _cpydiff_%s:\n\n' % output.name.rsplit('.', 1)[0])
- rst.write(output.desc + '\n')
- rst.write('~' * len(output.desc) + '\n\n')
- if output.cause != 'Unknown':
- rst.write('**Cause:** ' + output.cause + '\n\n')
- if output.workaround != 'Unknown':
- rst.write('**Workaround:** ' + output.workaround + '\n\n')
-
- rst.write('Sample code::\n\n' + indent(output.code, TAB) + '\n')
- output_cpy = indent(''.join(output.output_cpy[0:2]), TAB).rstrip()
- output_cpy = ('::\n\n' if output_cpy != '' else '') + output_cpy
- output_upy = indent(''.join(output.output_upy[0:2]), TAB).rstrip()
- output_upy = ('::\n\n' if output_upy != '' else '') + output_upy
- table = gen_table([['CPy output:', output_cpy], ['uPy output:', output_upy]])
+ rst.write(".. _cpydiff_%s:\n\n" % output.name.rsplit(".", 1)[0])
+ rst.write(output.desc + "\n")
+ rst.write("~" * len(output.desc) + "\n\n")
+ if output.cause != "Unknown":
+ rst.write("**Cause:** " + output.cause + "\n\n")
+ if output.workaround != "Unknown":
+ rst.write("**Workaround:** " + output.workaround + "\n\n")
+
+ rst.write("Sample code::\n\n" + indent(output.code, TAB) + "\n")
+ output_cpy = indent("".join(output.output_cpy[0:2]), TAB).rstrip()
+ output_cpy = ("::\n\n" if output_cpy != "" else "") + output_cpy
+ output_upy = indent("".join(output.output_upy[0:2]), TAB).rstrip()
+ output_upy = ("::\n\n" if output_upy != "" else "") + output_upy
+ table = gen_table([["CPy output:", output_cpy], ["uPy output:", output_upy]])
rst.write(table)
- template = open(INDEXTEMPLATE, 'r')
- index = open(DOCPATH + INDEX, 'w')
+ template = open(INDEXTEMPLATE, "r")
+ index = open(DOCPATH + INDEX, "w")
index.write(HEADER)
index.write(template.read())
for section in INDEXPRIORITY:
if section in toctree:
- index.write(indent(section + '.rst', TAB))
+ index.write(indent(section + ".rst", TAB))
toctree.remove(section)
for section in toctree:
- index.write(indent(section + '.rst', TAB))
+ index.write(indent(section + ".rst", TAB))
+
def main():
""" Main function """
# set search path so that test scripts find the test modules (and no other ones)
- os.environ['PYTHONPATH'] = TESTPATH
- os.environ['MICROPYPATH'] = TESTPATH
+ os.environ["PYTHONPATH"] = TESTPATH
+ os.environ["MICROPYPATH"] = TESTPATH
files = readfiles()
results = run_tests(files)
gen_rst(results)
+
main()
diff --git a/tools/gendoc.py b/tools/gendoc.py
index 61844d203..0f1305699 100644
--- a/tools/gendoc.py
+++ b/tools/gendoc.py
@@ -15,10 +15,12 @@ def re_match_first(regexs, line):
return name, match
return None, None
+
def makedirs(d):
if not os.path.isdir(d):
os.makedirs(d)
+
class Lexer:
class LexerError(Exception):
pass
@@ -31,15 +33,15 @@ class Lexer:
def __init__(self, file):
self.filename = file
- with open(file, 'rt') as f:
+ with open(file, "rt") as f:
line_num = 0
lines = []
for line in f:
line_num += 1
line = line.strip()
- if line == '///':
- lines.append((line_num, ''))
- elif line.startswith('/// '):
+ if line == "///":
+ lines.append((line_num, ""))
+ elif line.startswith("/// "):
lines.append((line_num, line[4:]))
elif len(lines) > 0 and lines[-1][1] is not None:
lines.append((line_num, None))
@@ -64,9 +66,10 @@ class Lexer:
return l[1]
def error(self, msg):
- print('({}:{}) {}'.format(self.filename, self.cur_line, msg))
+ print("({}:{}) {}".format(self.filename, self.cur_line, msg))
raise Lexer.LexerError
+
class MarkdownWriter:
def __init__(self):
pass
@@ -75,52 +78,53 @@ class MarkdownWriter:
self.lines = []
def end(self):
- return '\n'.join(self.lines)
+ return "\n".join(self.lines)
def heading(self, level, text):
if len(self.lines) > 0:
- self.lines.append('')
- self.lines.append(level * '#' + ' ' + text)
- self.lines.append('')
+ self.lines.append("")
+ self.lines.append(level * "#" + " " + text)
+ self.lines.append("")
def para(self, text):
- if len(self.lines) > 0 and self.lines[-1] != '':
- self.lines.append('')
+ if len(self.lines) > 0 and self.lines[-1] != "":
+ self.lines.append("")
if isinstance(text, list):
self.lines.extend(text)
elif isinstance(text, str):
self.lines.append(text)
else:
assert False
- self.lines.append('')
+ self.lines.append("")
def single_line(self, text):
self.lines.append(text)
def module(self, name, short_descr, descr):
- self.heading(1, 'module {}'.format(name))
+ self.heading(1, "module {}".format(name))
self.para(descr)
def function(self, ctx, name, args, descr):
- proto = '{}.{}{}'.format(ctx, self.name, self.args)
- self.heading(3, '`' + proto + '`')
+ proto = "{}.{}{}".format(ctx, self.name, self.args)
+ self.heading(3, "`" + proto + "`")
self.para(descr)
def method(self, ctx, name, args, descr):
- if name == '\\constructor':
- proto = '{}{}'.format(ctx, args)
- elif name == '\\call':
- proto = '{}{}'.format(ctx, args)
+ if name == "\\constructor":
+ proto = "{}{}".format(ctx, args)
+ elif name == "\\call":
+ proto = "{}{}".format(ctx, args)
else:
- proto = '{}.{}{}'.format(ctx, name, args)
- self.heading(3, '`' + proto + '`')
+ proto = "{}.{}{}".format(ctx, name, args)
+ self.heading(3, "`" + proto + "`")
self.para(descr)
def constant(self, ctx, name, descr):
- self.single_line('`{}.{}` - {}'.format(ctx, name, descr))
+ self.single_line("`{}.{}` - {}".format(ctx, name, descr))
+
class ReStructuredTextWriter:
- head_chars = {1:'=', 2:'-', 3:'.'}
+ head_chars = {1: "=", 2: "-", 3: "."}
def __init__(self):
pass
@@ -129,23 +133,23 @@ class ReStructuredTextWriter:
self.lines = []
def end(self):
- return '\n'.join(self.lines)
+ return "\n".join(self.lines)
def _convert(self, text):
- return text.replace('`', '``').replace('*', '\\*')
+ return text.replace("`", "``").replace("*", "\\*")
def heading(self, level, text, convert=True):
if len(self.lines) > 0:
- self.lines.append('')
+ self.lines.append("")
if convert:
text = self._convert(text)
self.lines.append(text)
self.lines.append(len(text) * self.head_chars[level])
- self.lines.append('')
+ self.lines.append("")
- def para(self, text, indent=''):
- if len(self.lines) > 0 and self.lines[-1] != '':
- self.lines.append('')
+ def para(self, text, indent=""):
+ if len(self.lines) > 0 and self.lines[-1] != "":
+ self.lines.append("")
if isinstance(text, list):
for t in text:
self.lines.append(indent + self._convert(t))
@@ -153,39 +157,41 @@ class ReStructuredTextWriter:
self.lines.append(indent + self._convert(text))
else:
assert False
- self.lines.append('')
+ self.lines.append("")
def single_line(self, text):
self.lines.append(self._convert(text))
def module(self, name, short_descr, descr):
- self.heading(1, ':mod:`{}` --- {}'.format(name, self._convert(short_descr)), convert=False)
- self.lines.append('.. module:: {}'.format(name))
- self.lines.append(' :synopsis: {}'.format(short_descr))
+ self.heading(1, ":mod:`{}` --- {}".format(name, self._convert(short_descr)), convert=False)
+ self.lines.append(".. module:: {}".format(name))
+ self.lines.append(" :synopsis: {}".format(short_descr))
self.para(descr)
def function(self, ctx, name, args, descr):
args = self._convert(args)
- self.lines.append('.. function:: ' + name + args)
- self.para(descr, indent=' ')
+ self.lines.append(".. function:: " + name + args)
+ self.para(descr, indent=" ")
def method(self, ctx, name, args, descr):
args = self._convert(args)
- if name == '\\constructor':
- self.lines.append('.. class:: ' + ctx + args)
- elif name == '\\call':
- self.lines.append('.. method:: ' + ctx + args)
+ if name == "\\constructor":
+ self.lines.append(".. class:: " + ctx + args)
+ elif name == "\\call":
+ self.lines.append(".. method:: " + ctx + args)
else:
- self.lines.append('.. method:: ' + ctx + '.' + name + args)
- self.para(descr, indent=' ')
+ self.lines.append(".. method:: " + ctx + "." + name + args)
+ self.para(descr, indent=" ")
def constant(self, ctx, name, descr):
- self.lines.append('.. data:: ' + name)
- self.para(descr, indent=' ')
+ self.lines.append(".. data:: " + name)
+ self.para(descr, indent=" ")
+
class DocValidateError(Exception):
pass
+
class DocItem:
def __init__(self):
self.doc = []
@@ -202,6 +208,7 @@ class DocItem:
def dump(self, writer):
writer.para(self.doc)
+
class DocConstant(DocItem):
def __init__(self, name, descr):
super().__init__()
@@ -211,6 +218,7 @@ class DocConstant(DocItem):
def dump(self, ctx, writer):
writer.constant(ctx, self.name, self.descr)
+
class DocFunction(DocItem):
def __init__(self, name, args):
super().__init__()
@@ -220,6 +228,7 @@ class DocFunction(DocItem):
def dump(self, ctx, writer):
writer.function(ctx, self.name, self.args, self.doc)
+
class DocMethod(DocItem):
def __init__(self, name, args):
super().__init__()
@@ -229,6 +238,7 @@ class DocMethod(DocItem):
def dump(self, ctx, writer):
writer.method(ctx, self.name, self.args, self.doc)
+
class DocClass(DocItem):
def __init__(self, name, descr):
super().__init__()
@@ -240,51 +250,52 @@ class DocClass(DocItem):
self.constants = {}
def process_classmethod(self, lex, d):
- name = d['id']
- if name == '\\constructor':
+ name = d["id"]
+ if name == "\\constructor":
dict_ = self.constructors
else:
dict_ = self.classmethods
if name in dict_:
lex.error("multiple definition of method '{}'".format(name))
- method = dict_[name] = DocMethod(name, d['args'])
+ method = dict_[name] = DocMethod(name, d["args"])
method.add_doc(lex)
def process_method(self, lex, d):
- name = d['id']
+ name = d["id"]
dict_ = self.methods
if name in dict_:
lex.error("multiple definition of method '{}'".format(name))
- method = dict_[name] = DocMethod(name, d['args'])
+ method = dict_[name] = DocMethod(name, d["args"])
method.add_doc(lex)
def process_constant(self, lex, d):
- name = d['id']
+ name = d["id"]
if name in self.constants:
lex.error("multiple definition of constant '{}'".format(name))
- self.constants[name] = DocConstant(name, d['descr'])
+ self.constants[name] = DocConstant(name, d["descr"])
lex.opt_break()
def dump(self, writer):
- writer.heading(1, 'class {}'.format(self.name))
+ writer.heading(1, "class {}".format(self.name))
super().dump(writer)
if len(self.constructors) > 0:
- writer.heading(2, 'Constructors')
- for f in sorted(self.constructors.values(), key=lambda x:x.name):
+ writer.heading(2, "Constructors")
+ for f in sorted(self.constructors.values(), key=lambda x: x.name):
f.dump(self.name, writer)
if len(self.classmethods) > 0:
- writer.heading(2, 'Class methods')
- for f in sorted(self.classmethods.values(), key=lambda x:x.name):
+ writer.heading(2, "Class methods")
+ for f in sorted(self.classmethods.values(), key=lambda x: x.name):
f.dump(self.name, writer)
if len(self.methods) > 0:
- writer.heading(2, 'Methods')
- for f in sorted(self.methods.values(), key=lambda x:x.name):
+ writer.heading(2, "Methods")
+ for f in sorted(self.methods.values(), key=lambda x: x.name):
f.dump(self.name.lower(), writer)
if len(self.constants) > 0:
- writer.heading(2, 'Constants')
- for c in sorted(self.constants.values(), key=lambda x:x.name):
+ writer.heading(2, "Constants")
+ for c in sorted(self.constants.values(), key=lambda x: x.name):
c.dump(self.name, writer)
+
class DocModule(DocItem):
def __init__(self, name, descr):
super().__init__()
@@ -299,22 +310,22 @@ class DocModule(DocItem):
self.cur_class = None
def process_function(self, lex, d):
- name = d['id']
+ name = d["id"]
if name in self.functions:
lex.error("multiple definition of function '{}'".format(name))
- function = self.functions[name] = DocFunction(name, d['args'])
+ function = self.functions[name] = DocFunction(name, d["args"])
function.add_doc(lex)
- #def process_classref(self, lex, d):
+ # def process_classref(self, lex, d):
# name = d['id']
# self.classes[name] = name
# lex.opt_break()
def process_class(self, lex, d):
- name = d['id']
+ name = d["id"]
if name in self.classes:
lex.error("multiple definition of class '{}'".format(name))
- self.cur_class = self.classes[name] = DocClass(name, d['descr'])
+ self.cur_class = self.classes[name] = DocClass(name, d["descr"])
self.cur_class.add_doc(lex)
def process_classmethod(self, lex, d):
@@ -326,10 +337,10 @@ class DocModule(DocItem):
def process_constant(self, lex, d):
if self.cur_class is None:
# a module-level constant
- name = d['id']
+ name = d["id"]
if name in self.constants:
lex.error("multiple definition of constant '{}'".format(name))
- self.constants[name] = DocConstant(name, d['descr'])
+ self.constants[name] = DocConstant(name, d["descr"])
lex.opt_break()
else:
# a class-level constant
@@ -337,50 +348,51 @@ class DocModule(DocItem):
def validate(self):
if self.descr is None:
- raise DocValidateError('module {} referenced but never defined'.format(self.name))
+ raise DocValidateError("module {} referenced but never defined".format(self.name))
def dump(self, writer):
writer.module(self.name, self.descr, self.doc)
if self.functions:
- writer.heading(2, 'Functions')
- for f in sorted(self.functions.values(), key=lambda x:x.name):
+ writer.heading(2, "Functions")
+ for f in sorted(self.functions.values(), key=lambda x: x.name):
f.dump(self.name, writer)
if self.constants:
- writer.heading(2, 'Constants')
- for c in sorted(self.constants.values(), key=lambda x:x.name):
+ writer.heading(2, "Constants")
+ for c in sorted(self.constants.values(), key=lambda x: x.name):
c.dump(self.name, writer)
if self.classes:
- writer.heading(2, 'Classes')
- for c in sorted(self.classes.values(), key=lambda x:x.name):
- writer.para('[`{}.{}`]({}) - {}'.format(self.name, c.name, c.name, c.descr))
+ writer.heading(2, "Classes")
+ for c in sorted(self.classes.values(), key=lambda x: x.name):
+ writer.para("[`{}.{}`]({}) - {}".format(self.name, c.name, c.name, c.descr))
def write_html(self, dir):
md_writer = MarkdownWriter()
md_writer.start()
self.dump(md_writer)
- with open(os.path.join(dir, 'index.html'), 'wt') as f:
+ with open(os.path.join(dir, "index.html"), "wt") as f:
f.write(markdown.markdown(md_writer.end()))
for c in self.classes.values():
class_dir = os.path.join(dir, c.name)
makedirs(class_dir)
md_writer.start()
- md_writer.para('part of the [{} module](./)'.format(self.name))
+ md_writer.para("part of the [{} module](./)".format(self.name))
c.dump(md_writer)
- with open(os.path.join(class_dir, 'index.html'), 'wt') as f:
+ with open(os.path.join(class_dir, "index.html"), "wt") as f:
f.write(markdown.markdown(md_writer.end()))
def write_rst(self, dir):
rst_writer = ReStructuredTextWriter()
rst_writer.start()
self.dump(rst_writer)
- with open(dir + '/' + self.name + '.rst', 'wt') as f:
+ with open(dir + "/" + self.name + ".rst", "wt") as f:
f.write(rst_writer.end())
for c in self.classes.values():
rst_writer.start()
c.dump(rst_writer)
- with open(dir + '/' + self.name + '.' + c.name + '.rst', 'wt') as f:
+ with open(dir + "/" + self.name + "." + c.name + ".rst", "wt") as f:
f.write(rst_writer.end())
+
class Doc:
def __init__(self):
self.modules = {}
@@ -393,20 +405,20 @@ class Doc:
def check_module(self, lex):
if self.cur_module is None:
- lex.error('module not defined')
+ lex.error("module not defined")
def process_module(self, lex, d):
- name = d['id']
+ name = d["id"]
if name not in self.modules:
self.modules[name] = DocModule(name, None)
self.cur_module = self.modules[name]
if self.cur_module.descr is not None:
lex.error("multiple definition of module '{}'".format(name))
- self.cur_module.descr = d['descr']
+ self.cur_module.descr = d["descr"]
self.cur_module.add_doc(lex)
def process_moduleref(self, lex, d):
- name = d['id']
+ name = d["id"]
if name not in self.modules:
self.modules[name] = DocModule(name, None)
self.cur_module = self.modules[name]
@@ -437,41 +449,46 @@ class Doc:
m.validate()
def dump(self, writer):
- writer.heading(1, 'Modules')
- writer.para('These are the Python modules that are implemented.')
- for m in sorted(self.modules.values(), key=lambda x:x.name):
- writer.para('[`{}`]({}/) - {}'.format(m.name, m.name, m.descr))
+ writer.heading(1, "Modules")
+ writer.para("These are the Python modules that are implemented.")
+ for m in sorted(self.modules.values(), key=lambda x: x.name):
+ writer.para("[`{}`]({}/) - {}".format(m.name, m.name, m.descr))
def write_html(self, dir):
md_writer = MarkdownWriter()
- with open(os.path.join(dir, 'module', 'index.html'), 'wt') as f:
+ with open(os.path.join(dir, "module", "index.html"), "wt") as f:
md_writer.start()
self.dump(md_writer)
f.write(markdown.markdown(md_writer.end()))
for m in self.modules.values():
- mod_dir = os.path.join(dir, 'module', m.name)
+ mod_dir = os.path.join(dir, "module", m.name)
makedirs(mod_dir)
m.write_html(mod_dir)
def write_rst(self, dir):
- #with open(os.path.join(dir, 'module', 'index.html'), 'wt') as f:
+ # with open(os.path.join(dir, 'module', 'index.html'), 'wt') as f:
# f.write(markdown.markdown(self.dump()))
for m in self.modules.values():
m.write_rst(dir)
-regex_descr = r'(?P<descr>.*)'
+
+regex_descr = r"(?P<descr>.*)"
doc_regexs = (
- (Doc.process_module, re.compile(r'\\module (?P<id>[a-z][a-z0-9]*) - ' + regex_descr + r'$')),
- (Doc.process_moduleref, re.compile(r'\\moduleref (?P<id>[a-z]+)$')),
- (Doc.process_function, re.compile(r'\\function (?P<id>[a-z0-9_]+)(?P<args>\(.*\))$')),
- (Doc.process_classmethod, re.compile(r'\\classmethod (?P<id>\\?[a-z0-9_]+)(?P<args>\(.*\))$')),
- (Doc.process_method, re.compile(r'\\method (?P<id>\\?[a-z0-9_]+)(?P<args>\(.*\))$')),
- (Doc.process_constant, re.compile(r'\\constant (?P<id>[A-Za-z0-9_]+) - ' + regex_descr + r'$')),
- #(Doc.process_classref, re.compile(r'\\classref (?P<id>[A-Za-z0-9_]+)$')),
- (Doc.process_class, re.compile(r'\\class (?P<id>[A-Za-z0-9_]+) - ' + regex_descr + r'$')),
+ (Doc.process_module, re.compile(r"\\module (?P<id>[a-z][a-z0-9]*) - " + regex_descr + r"$")),
+ (Doc.process_moduleref, re.compile(r"\\moduleref (?P<id>[a-z]+)$")),
+ (Doc.process_function, re.compile(r"\\function (?P<id>[a-z0-9_]+)(?P<args>\(.*\))$")),
+ (Doc.process_classmethod, re.compile(r"\\classmethod (?P<id>\\?[a-z0-9_]+)(?P<args>\(.*\))$")),
+ (Doc.process_method, re.compile(r"\\method (?P<id>\\?[a-z0-9_]+)(?P<args>\(.*\))$")),
+ (
+ Doc.process_constant,
+ re.compile(r"\\constant (?P<id>[A-Za-z0-9_]+) - " + regex_descr + r"$"),
+ ),
+ # (Doc.process_classref, re.compile(r'\\classref (?P<id>[A-Za-z0-9_]+)$')),
+ (Doc.process_class, re.compile(r"\\class (?P<id>[A-Za-z0-9_]+) - " + regex_descr + r"$")),
)
+
def process_file(file, doc):
lex = Lexer(file)
doc.new_file()
@@ -481,11 +498,11 @@ def process_file(file, doc):
line = lex.next()
fun, match = re_match_first(doc_regexs, line)
if fun == None:
- lex.error('unknown line format: {}'.format(line))
+ lex.error("unknown line format: {}".format(line))
fun(doc, lex, match.groupdict())
except Lexer.Break:
- lex.error('unexpected break')
+ lex.error("unexpected break")
except Lexer.EOF:
pass
@@ -495,16 +512,21 @@ def process_file(file, doc):
return True
+
def main():
- cmd_parser = argparse.ArgumentParser(description='Generate documentation for pyboard API from C files.')
- cmd_parser.add_argument('--outdir', metavar='<output dir>', default='gendoc-out', help='ouput directory')
- cmd_parser.add_argument('--format', default='html', help='output format: html or rst')
- cmd_parser.add_argument('files', nargs='+', help='input files')
+ cmd_parser = argparse.ArgumentParser(
+ description="Generate documentation for pyboard API from C files."
+ )
+ cmd_parser.add_argument(
+ "--outdir", metavar="<output dir>", default="gendoc-out", help="ouput directory"
+ )
+ cmd_parser.add_argument("--format", default="html", help="output format: html or rst")
+ cmd_parser.add_argument("files", nargs="+", help="input files")
args = cmd_parser.parse_args()
doc = Doc()
for file in args.files:
- print('processing', file)
+ print("processing", file)
if not process_file(file, doc):
return
try:
@@ -514,15 +536,16 @@ def main():
makedirs(args.outdir)
- if args.format == 'html':
+ if args.format == "html":
doc.write_html(args.outdir)
- elif args.format == 'rst':
+ elif args.format == "rst":
doc.write_rst(args.outdir)
else:
- print('unknown format:', args.format)
+ print("unknown format:", args.format)
return
- print('written to', args.outdir)
+ print("written to", args.outdir)
+
if __name__ == "__main__":
main()
diff --git a/tools/insert-usb-ids.py b/tools/insert-usb-ids.py
index cdccd3be9..b1d848ad4 100644
--- a/tools/insert-usb-ids.py
+++ b/tools/insert-usb-ids.py
@@ -8,15 +8,16 @@ import sys
import re
import string
-needed_keys = ('USB_PID_CDC_MSC', 'USB_PID_CDC_HID', 'USB_PID_CDC', 'USB_VID')
+needed_keys = ("USB_PID_CDC_MSC", "USB_PID_CDC_HID", "USB_PID_CDC", "USB_VID")
+
def parse_usb_ids(filename):
rv = dict()
for line in open(filename).readlines():
- line = line.rstrip('\r\n')
- match = re.match('^#define\s+(\w+)\s+\(0x([0-9A-Fa-f]+)\)$', line)
- if match and match.group(1).startswith('USBD_'):
- key = match.group(1).replace('USBD', 'USB')
+ line = line.rstrip("\r\n")
+ match = re.match("^#define\s+(\w+)\s+\(0x([0-9A-Fa-f]+)\)$", line)
+ if match and match.group(1).startswith("USBD_"):
+ key = match.group(1).replace("USBD", "USB")
val = match.group(2)
print("key =", key, "val =", val)
if key in needed_keys:
@@ -26,9 +27,10 @@ def parse_usb_ids(filename):
raise Exception("Unable to parse %s from %s" % (k, filename))
return rv
+
if __name__ == "__main__":
usb_ids_file = sys.argv[1]
template_file = sys.argv[2]
replacements = parse_usb_ids(usb_ids_file)
- for line in open(template_file, 'r').readlines():
- print(string.Template(line).safe_substitute(replacements), end='')
+ for line in open(template_file, "r").readlines():
+ print(string.Template(line).safe_substitute(replacements), end="")
diff --git a/tools/make-frozen.py b/tools/make-frozen.py
index 8809fd0c8..a8706bf78 100755
--- a/tools/make-frozen.py
+++ b/tools/make-frozen.py
@@ -25,6 +25,7 @@ import os
def module_name(f):
return f
+
modules = []
if len(sys.argv) > 1:
@@ -35,7 +36,7 @@ if len(sys.argv) > 1:
for f in filenames:
fullpath = dirpath + "/" + f
st = os.stat(fullpath)
- modules.append((fullpath[root_len + 1:], st))
+ modules.append((fullpath[root_len + 1 :], st))
print("#include <stdint.h>")
print("const char mp_frozen_str_names[] = {")
@@ -62,8 +63,8 @@ for f, st in modules:
# data. We could just encode all characters as hex digits but it's nice
# to be able to read the resulting C code as ASCII when possible.
- data = bytearray(data) # so Python2 extracts each byte as an integer
- esc_dict = {ord('\n'): '\\n', ord('\r'): '\\r', ord('"'): '\\"', ord('\\'): '\\\\'}
+ data = bytearray(data) # so Python2 extracts each byte as an integer
+ esc_dict = {ord("\n"): "\\n", ord("\r"): "\\r", ord('"'): '\\"', ord("\\"): "\\\\"}
chrs = ['"']
break_str = False
for c in data:
@@ -76,9 +77,9 @@ for f, st in modules:
break_str = False
chrs.append(chr(c))
else:
- chrs.append('\\x%02x' % c)
+ chrs.append("\\x%02x" % c)
break_str = True
chrs.append('\\0"')
- print(''.join(chrs))
+ print("".join(chrs))
print("};")
diff --git a/tools/makemanifest.py b/tools/makemanifest.py
index 2caf7b0db..a8b310ad6 100644
--- a/tools/makemanifest.py
+++ b/tools/makemanifest.py
@@ -33,6 +33,7 @@ import subprocess
###########################################################################
# Public functions to be used in the manifest
+
def include(manifest):
"""Include another manifest.
@@ -55,6 +56,7 @@ def include(manifest):
exec(f.read())
os.chdir(prev_cwd)
+
def freeze(path, script=None, opt=0):
"""Freeze the input, automatically determining its type. A .py script
will be compiled to a .mpy first then frozen, and a .mpy file will be
@@ -84,6 +86,7 @@ def freeze(path, script=None, opt=0):
freeze_internal(KIND_AUTO, path, script, opt)
+
def freeze_as_str(path):
"""Freeze the given `path` and all .py scripts within it as a string,
which will be compiled upon import.
@@ -91,6 +94,7 @@ def freeze_as_str(path):
freeze_internal(KIND_AS_STR, path, None, 0)
+
def freeze_as_mpy(path, script=None, opt=0):
"""Freeze the input (see above) by first compiling the .py scripts to
.mpy files, then freezing the resulting .mpy files.
@@ -98,6 +102,7 @@ def freeze_as_mpy(path, script=None, opt=0):
freeze_internal(KIND_AS_MPY, path, script, opt)
+
def freeze_mpy(path, script=None, opt=0):
"""Freeze the input (see above), which must be .mpy files that are
frozen directly.
@@ -118,9 +123,11 @@ VARS = {}
manifest_list = []
+
class FreezeError(Exception):
pass
+
def system(cmd):
try:
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
@@ -128,23 +135,26 @@ def system(cmd):
except subprocess.CalledProcessError as er:
return -1, er.output
+
def convert_path(path):
# Perform variable substituion.
for name, value in VARS.items():
- path = path.replace('$({})'.format(name), value)
+ path = path.replace("$({})".format(name), value)
# Convert to absolute path (so that future operations don't rely on
# still being chdir'ed).
return os.path.abspath(path)
+
def get_timestamp(path, default=None):
try:
stat = os.stat(path)
return stat.st_mtime
except OSError:
if default is None:
- raise FreezeError('cannot stat {}'.format(path))
+ raise FreezeError("cannot stat {}".format(path))
return default
+
def get_timestamp_newest(path):
ts_newest = 0
for dirpath, dirnames, filenames in os.walk(path, followlinks=True):
@@ -152,82 +162,90 @@ def get_timestamp_newest(path):
ts_newest = max(ts_newest, get_timestamp(os.path.join(dirpath, f)))
return ts_newest
+
def mkdir(path):
- cur_path = ''
- for p in path.split('/')[:-1]:
- cur_path += p + '/'
+ cur_path = ""
+ for p in path.split("/")[:-1]:
+ cur_path += p + "/"
try:
os.mkdir(cur_path)
except OSError as er:
- if er.args[0] == 17: # file exists
+ if er.args[0] == 17: # file exists
pass
else:
raise er
+
def freeze_internal(kind, path, script, opt):
path = convert_path(path)
if script is None and kind == KIND_AS_STR:
if any(f[0] == KIND_AS_STR for f in manifest_list):
- raise FreezeError('can only freeze one str directory')
+ raise FreezeError("can only freeze one str directory")
manifest_list.append((KIND_AS_STR, path, script, opt))
elif script is None:
for dirpath, dirnames, filenames in os.walk(path, followlinks=True):
for f in filenames:
- freeze_internal(kind, path, (dirpath + '/' + f)[len(path) + 1:], opt)
+ freeze_internal(kind, path, (dirpath + "/" + f)[len(path) + 1 :], opt)
elif not isinstance(script, str):
for s in script:
freeze_internal(kind, path, s, opt)
else:
- extension_kind = {KIND_AS_MPY: '.py', KIND_MPY: '.mpy'}
+ extension_kind = {KIND_AS_MPY: ".py", KIND_MPY: ".mpy"}
if kind == KIND_AUTO:
for k, ext in extension_kind.items():
if script.endswith(ext):
kind = k
break
else:
- print('warn: unsupported file type, skipped freeze: {}'.format(script))
+ print("warn: unsupported file type, skipped freeze: {}".format(script))
return
wanted_extension = extension_kind[kind]
if not script.endswith(wanted_extension):
- raise FreezeError('expecting a {} file, got {}'.format(wanted_extension, script))
+ raise FreezeError("expecting a {} file, got {}".format(wanted_extension, script))
manifest_list.append((kind, path, script, opt))
+
def main():
# Parse arguments
import argparse
- cmd_parser = argparse.ArgumentParser(description='A tool to generate frozen content in MicroPython firmware images.')
- cmd_parser.add_argument('-o', '--output', help='output path')
- cmd_parser.add_argument('-b', '--build-dir', help='output path')
- cmd_parser.add_argument('-f', '--mpy-cross-flags', default='', help='flags to pass to mpy-cross')
- cmd_parser.add_argument('-v', '--var', action='append', help='variables to substitute')
- cmd_parser.add_argument('files', nargs='+', help='input manifest list')
+
+ cmd_parser = argparse.ArgumentParser(
+ description="A tool to generate frozen content in MicroPython firmware images."
+ )
+ cmd_parser.add_argument("-o", "--output", help="output path")
+ cmd_parser.add_argument("-b", "--build-dir", help="output path")
+ cmd_parser.add_argument(
+ "-f", "--mpy-cross-flags", default="", help="flags to pass to mpy-cross"
+ )
+ cmd_parser.add_argument("-v", "--var", action="append", help="variables to substitute")
+ cmd_parser.add_argument("files", nargs="+", help="input manifest list")
args = cmd_parser.parse_args()
# Extract variables for substitution.
for var in args.var:
- name, value = var.split('=', 1)
+ name, value = var.split("=", 1)
if os.path.exists(value):
value = os.path.abspath(value)
VARS[name] = value
- if 'MPY_DIR' not in VARS or 'PORT_DIR' not in VARS:
- print('MPY_DIR and PORT_DIR variables must be specified')
+ if "MPY_DIR" not in VARS or "PORT_DIR" not in VARS:
+ print("MPY_DIR and PORT_DIR variables must be specified")
sys.exit(1)
# Get paths to tools
- MAKE_FROZEN = VARS['MPY_DIR'] + '/tools/make-frozen.py'
- MPY_CROSS = VARS['MPY_DIR'] + '/mpy-cross/mpy-cross'
- MPY_TOOL = VARS['MPY_DIR'] + '/tools/mpy-tool.py'
+ MAKE_FROZEN = VARS["MPY_DIR"] + "/tools/make-frozen.py"
+ MPY_CROSS = VARS["MPY_DIR"] + "/mpy-cross/mpy-cross"
+ MPY_TOOL = VARS["MPY_DIR"] + "/tools/mpy-tool.py"
# Ensure mpy-cross is built
if not os.path.exists(MPY_CROSS):
- print('mpy-cross not found at {}, please build it first'.format(MPY_CROSS))
+ print("mpy-cross not found at {}, please build it first".format(MPY_CROSS))
sys.exit(1)
# Include top-level inputs, to generate the manifest
for input_manifest in args.files:
try:
- if input_manifest.endswith('.py'):
+ if input_manifest.endswith(".py"):
include(input_manifest)
else:
exec(input_manifest)
@@ -244,22 +262,26 @@ def main():
str_paths.append(path)
ts_outfile = get_timestamp_newest(path)
elif kind == KIND_AS_MPY:
- infile = '{}/{}'.format(path, script)
- outfile = '{}/frozen_mpy/{}.mpy'.format(args.build_dir, script[:-3])
+ infile = "{}/{}".format(path, script)
+ outfile = "{}/frozen_mpy/{}.mpy".format(args.build_dir, script[:-3])
ts_infile = get_timestamp(infile)
ts_outfile = get_timestamp(outfile, 0)
if ts_infile >= ts_outfile:
- print('MPY', script)
+ print("MPY", script)
mkdir(outfile)
- res, out = system([MPY_CROSS] + args.mpy_cross_flags.split() + ['-o', outfile, '-s', script, '-O{}'.format(opt), infile])
+ res, out = system(
+ [MPY_CROSS]
+ + args.mpy_cross_flags.split()
+ + ["-o", outfile, "-s", script, "-O{}".format(opt), infile]
+ )
if res != 0:
- print('error compiling {}: {}'.format(infile, out))
+ print("error compiling {}: {}".format(infile, out))
raise SystemExit(1)
ts_outfile = get_timestamp(outfile)
mpy_files.append(outfile)
else:
assert kind == KIND_MPY
- infile = '{}/{}'.format(path, script)
+ infile = "{}/{}".format(path, script)
mpy_files.append(infile)
ts_outfile = get_timestamp(infile)
ts_newest = max(ts_newest, ts_outfile)
@@ -272,35 +294,45 @@ def main():
# Freeze paths as strings
res, output_str = system([sys.executable, MAKE_FROZEN] + str_paths)
if res != 0:
- print('error freezing strings {}: {}'.format(str_paths, output_str))
+ print("error freezing strings {}: {}".format(str_paths, output_str))
sys.exit(1)
# Freeze .mpy files
if mpy_files:
- res, output_mpy = system([sys.executable, MPY_TOOL, '-f', '-q', args.build_dir + '/genhdr/qstrdefs.preprocessed.h'] + mpy_files)
+ res, output_mpy = system(
+ [
+ sys.executable,
+ MPY_TOOL,
+ "-f",
+ "-q",
+ args.build_dir + "/genhdr/qstrdefs.preprocessed.h",
+ ]
+ + mpy_files
+ )
if res != 0:
- print('error freezing mpy {}:'.format(mpy_files))
- print(str(output_mpy, 'utf8'))
+ print("error freezing mpy {}:".format(mpy_files))
+ print(str(output_mpy, "utf8"))
sys.exit(1)
else:
output_mpy = (
b'#include "py/emitglue.h"\n'
- b'extern const qstr_pool_t mp_qstr_const_pool;\n'
- b'const qstr_pool_t mp_qstr_frozen_const_pool = {\n'
- b' (qstr_pool_t*)&mp_qstr_const_pool, MP_QSTRnumber_of, 0, 0\n'
- b'};\n'
+ b"extern const qstr_pool_t mp_qstr_const_pool;\n"
+ b"const qstr_pool_t mp_qstr_frozen_const_pool = {\n"
+ b" (qstr_pool_t*)&mp_qstr_const_pool, MP_QSTRnumber_of, 0, 0\n"
+ b"};\n"
b'const char mp_frozen_mpy_names[1] = {"\\0"};\n'
- b'const mp_raw_code_t *const mp_frozen_mpy_content[0] = {};\n'
+ b"const mp_raw_code_t *const mp_frozen_mpy_content[0] = {};\n"
)
# Generate output
- print('GEN', args.output)
+ print("GEN", args.output)
mkdir(args.output)
- with open(args.output, 'wb') as f:
- f.write(b'//\n// Content for MICROPY_MODULE_FROZEN_STR\n//\n')
+ with open(args.output, "wb") as f:
+ f.write(b"//\n// Content for MICROPY_MODULE_FROZEN_STR\n//\n")
f.write(output_str)
- f.write(b'//\n// Content for MICROPY_MODULE_FROZEN_MPY\n//\n')
+ f.write(b"//\n// Content for MICROPY_MODULE_FROZEN_MPY\n//\n")
f.write(output_mpy)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
diff --git a/tools/metrics.py b/tools/metrics.py
index 0cc6db510..5f1560287 100755
--- a/tools/metrics.py
+++ b/tools/metrics.py
@@ -45,7 +45,8 @@ Other commands:
import sys, re, subprocess
-MAKE_FLAGS = ['-j3', 'CFLAGS_EXTRA=-DNDEBUG']
+MAKE_FLAGS = ["-j3", "CFLAGS_EXTRA=-DNDEBUG"]
+
class PortData:
def __init__(self, name, dir, output, make_flags=None):
@@ -54,19 +55,21 @@ class PortData:
self.output = output
self.make_flags = make_flags
+
port_data = {
- 'b': PortData('bare-arm', 'bare-arm', 'build/firmware.elf'),
- 'm': PortData('minimal x86', 'minimal', 'build/firmware.elf'),
- 'u': PortData('unix x64', 'unix', 'micropython'),
- 'n': PortData('unix nanbox', 'unix', 'micropython-nanbox', 'VARIANT=nanbox'),
- 's': PortData('stm32', 'stm32', 'build-PYBV10/firmware.elf', 'BOARD=PYBV10'),
- 'c': PortData('cc3200', 'cc3200', 'build/WIPY/release/application.axf', 'BTARGET=application'),
- '8': PortData('esp8266', 'esp8266', 'build-GENERIC/firmware.elf'),
- '3': PortData('esp32', 'esp32', 'build-GENERIC/application.elf'),
- 'r': PortData('nrf', 'nrf', 'build-pca10040/firmware.elf'),
- 'd': PortData('samd', 'samd', 'build-ADAFRUIT_ITSYBITSY_M4_EXPRESS/firmware.elf'),
+ "b": PortData("bare-arm", "bare-arm", "build/firmware.elf"),
+ "m": PortData("minimal x86", "minimal", "build/firmware.elf"),
+ "u": PortData("unix x64", "unix", "micropython"),
+ "n": PortData("unix nanbox", "unix", "micropython-nanbox", "VARIANT=nanbox"),
+ "s": PortData("stm32", "stm32", "build-PYBV10/firmware.elf", "BOARD=PYBV10"),
+ "c": PortData("cc3200", "cc3200", "build/WIPY/release/application.axf", "BTARGET=application"),
+ "8": PortData("esp8266", "esp8266", "build-GENERIC/firmware.elf"),
+ "3": PortData("esp32", "esp32", "build-GENERIC/application.elf"),
+ "r": PortData("nrf", "nrf", "build-pca10040/firmware.elf"),
+ "d": PortData("samd", "samd", "build-ADAFRUIT_ITSYBITSY_M4_EXPRESS/firmware.elf"),
}
+
def syscmd(*args):
sys.stdout.flush()
a2 = []
@@ -77,6 +80,7 @@ def syscmd(*args):
a2.extend(a)
subprocess.run(a2)
+
def parse_port_list(args):
if not args:
return list(port_data.values())
@@ -87,10 +91,11 @@ def parse_port_list(args):
try:
ports.append(port_data[port_char])
except KeyError:
- print('unknown port:', port_char)
+ print("unknown port:", port_char)
sys.exit(1)
return ports
+
def read_build_log(filename):
data = dict()
lines = []
@@ -98,7 +103,7 @@ def read_build_log(filename):
with open(filename) as f:
for line in f:
line = line.strip()
- if line.strip() == 'COMPUTING SIZES':
+ if line.strip() == "COMPUTING SIZES":
found_sizes = True
elif found_sizes:
lines.append(line)
@@ -109,14 +114,15 @@ def read_build_log(filename):
data[fields[-1]] = [int(f) for f in fields[:-2]]
is_size_line = False
else:
- is_size_line = line.startswith('text\t ')
+ is_size_line = line.startswith("text\t ")
return data
+
def do_diff(args):
"""Compute the difference between firmware sizes."""
if len(args) != 2:
- print('usage: %s diff <out1> <out2>' % sys.argv[0])
+ print("usage: %s diff <out1> <out2>" % sys.argv[0])
sys.exit(1)
data1 = read_build_log(args[0])
@@ -125,81 +131,86 @@ def do_diff(args):
for key, value1 in data1.items():
value2 = data2[key]
for port in port_data.values():
- if key == 'ports/{}/{}'.format(port.dir, port.output):
+ if key == "ports/{}/{}".format(port.dir, port.output):
name = port.name
break
data = [v2 - v1 for v1, v2 in zip(value1, value2)]
- warn = ''
- board = re.search(r'/build-([A-Za-z0-9_]+)/', key)
+ warn = ""
+ board = re.search(r"/build-([A-Za-z0-9_]+)/", key)
if board:
board = board.group(1)
else:
- board = ''
- if name == 'cc3200':
+ board = ""
+ if name == "cc3200":
delta = data[0]
percent = 100 * delta / value1[0]
if data[1] != 0:
- warn += ' %+u(data)' % data[1]
+ warn += " %+u(data)" % data[1]
else:
delta = data[3]
percent = 100 * delta / value1[3]
if data[1] != 0:
- warn += ' %+u(data)' % data[1]
+ warn += " %+u(data)" % data[1]
if data[2] != 0:
- warn += ' %+u(bss)' % data[2]
+ warn += " %+u(bss)" % data[2]
if warn:
- warn = '[incl%s]' % warn
- print('%11s: %+5u %+.3f%% %s%s' % (name, delta, percent, board, warn))
+ warn = "[incl%s]" % warn
+ print("%11s: %+5u %+.3f%% %s%s" % (name, delta, percent, board, warn))
+
def do_clean(args):
"""Clean ports."""
ports = parse_port_list(args)
- print('CLEANING')
+ print("CLEANING")
for port in ports:
- syscmd('make', '-C', 'ports/{}'.format(port.dir), port.make_flags, 'clean')
+ syscmd("make", "-C", "ports/{}".format(port.dir), port.make_flags, "clean")
+
def do_build(args):
"""Build ports and print firmware sizes."""
ports = parse_port_list(args)
- print('BUILDING MPY-CROSS')
- syscmd('make', '-C', 'mpy-cross', MAKE_FLAGS)
+ print("BUILDING MPY-CROSS")
+ syscmd("make", "-C", "mpy-cross", MAKE_FLAGS)
- print('BUILDING PORTS')
+ print("BUILDING PORTS")
for port in ports:
- syscmd('make', '-C', 'ports/{}'.format(port.dir), MAKE_FLAGS, port.make_flags)
+ syscmd("make", "-C", "ports/{}".format(port.dir), MAKE_FLAGS, port.make_flags)
do_sizes(args)
+
def do_sizes(args):
"""Compute and print sizes of firmware."""
ports = parse_port_list(args)
- print('COMPUTING SIZES')
+ print("COMPUTING SIZES")
for port in ports:
- syscmd('size', 'ports/{}/{}'.format(port.dir, port.output))
+ syscmd("size", "ports/{}/{}".format(port.dir, port.output))
+
def main():
# Get command to execute
if len(sys.argv) == 1:
- print('Available commands:')
+ print("Available commands:")
for cmd in globals():
- if cmd.startswith('do_'):
- print(' {:9} {}'.format(cmd[3:], globals()[cmd].__doc__))
+ if cmd.startswith("do_"):
+ print(" {:9} {}".format(cmd[3:], globals()[cmd].__doc__))
sys.exit(1)
cmd = sys.argv.pop(1)
# Dispatch to desired command
try:
- cmd = globals()['do_{}'.format(cmd)]
+ cmd = globals()["do_{}".format(cmd)]
except KeyError:
print("{}: unknown command '{}'".format(sys.argv[0], cmd))
sys.exit(1)
cmd(sys.argv[1:])
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
diff --git a/tools/mpy-tool.py b/tools/mpy-tool.py
index 77a129944..0a9246503 100755
--- a/tools/mpy-tool.py
+++ b/tools/mpy-tool.py
@@ -27,7 +27,8 @@
# Python 2/3 compatibility code
from __future__ import print_function
import platform
-if platform.python_version_tuple()[0] == '2':
+
+if platform.python_version_tuple()[0] == "2":
str_cons = lambda val, enc=None: val
bytes_cons = lambda val, enc=None: bytearray(val)
is_str_type = lambda o: type(o) is str
@@ -45,48 +46,56 @@ import sys
import struct
from collections import namedtuple
-sys.path.append(sys.path[0] + '/../py')
+sys.path.append(sys.path[0] + "/../py")
import makeqstrdata as qstrutil
+
class FreezeError(Exception):
def __init__(self, rawcode, msg):
self.rawcode = rawcode
self.msg = msg
def __str__(self):
- return 'error while freezing %s: %s' % (self.rawcode.source_file, self.msg)
+ return "error while freezing %s: %s" % (self.rawcode.source_file, self.msg)
+
class Config:
MPY_VERSION = 5
MICROPY_LONGINT_IMPL_NONE = 0
MICROPY_LONGINT_IMPL_LONGLONG = 1
MICROPY_LONGINT_IMPL_MPZ = 2
+
+
config = Config()
+
class QStrType:
def __init__(self, str):
self.str = str
self.qstr_esc = qstrutil.qstr_escape(self.str)
- self.qstr_id = 'MP_QSTR_' + self.qstr_esc
+ self.qstr_id = "MP_QSTR_" + self.qstr_esc
+
# Initialise global list of qstrs with static qstrs
-global_qstrs = [None] # MP_QSTRnull should never be referenced
+global_qstrs = [None] # MP_QSTRnull should never be referenced
for n in qstrutil.static_qstr_list:
global_qstrs.append(QStrType(n))
+
class QStrWindow:
def __init__(self, size):
self.window = []
self.size = size
def push(self, val):
- self.window = [val] + self.window[:self.size - 1]
+ self.window = [val] + self.window[: self.size - 1]
def access(self, idx):
val = self.window[idx]
- self.window = [val] + self.window[:idx] + self.window[idx + 1:]
+ self.window = [val] + self.window[:idx] + self.window[idx + 1 :]
return val
+
MP_CODE_BYTECODE = 2
MP_CODE_NATIVE_PY = 3
MP_CODE_NATIVE_VIPER = 4
@@ -104,7 +113,7 @@ MP_NATIVE_ARCH_ARMV7EMDP = 8
MP_NATIVE_ARCH_XTENSA = 9
MP_NATIVE_ARCH_XTENSAWIN = 10
-MP_BC_MASK_EXTRA_BYTE = 0x9e
+MP_BC_MASK_EXTRA_BYTE = 0x9E
MP_BC_FORMAT_BYTE = 0
MP_BC_FORMAT_QSTR = 1
@@ -121,13 +130,15 @@ MP_BC_STORE_ATTR = 0x18
def mp_opcode_format(bytecode, ip, count_var_uint):
opcode = bytecode[ip]
ip_start = ip
- f = ((0x000003a4 >> (2 * ((opcode) >> 4))) & 3)
+ f = (0x000003A4 >> (2 * ((opcode) >> 4))) & 3
if f == MP_BC_FORMAT_QSTR:
if config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE:
- if (opcode == MP_BC_LOAD_NAME
+ if (
+ opcode == MP_BC_LOAD_NAME
or opcode == MP_BC_LOAD_GLOBAL
or opcode == MP_BC_LOAD_ATTR
- or opcode == MP_BC_STORE_ATTR):
+ or opcode == MP_BC_STORE_ATTR
+ ):
ip += 1
ip += 3
else:
@@ -143,10 +154,11 @@ def mp_opcode_format(bytecode, ip, count_var_uint):
ip += extra_byte
return f, ip - ip_start
+
def read_prelude_sig(read_byte):
z = read_byte()
# xSSSSEAA
- S = (z >> 3) & 0xf
+ S = (z >> 3) & 0xF
E = (z >> 2) & 0x1
F = 0
A = z & 0x3
@@ -166,6 +178,7 @@ def read_prelude_sig(read_byte):
S += 1
return S, E, F, A, K, D
+
def read_prelude_size(read_byte):
I = 0
C = 0
@@ -173,20 +186,29 @@ def read_prelude_size(read_byte):
while True:
z = read_byte()
# xIIIIIIC
- I |= ((z & 0x7e) >> 1) << (6 * n)
+ I |= ((z & 0x7E) >> 1) << (6 * n)
C |= (z & 1) << n
if not (z & 0x80):
break
n += 1
return I, C
+
def extract_prelude(bytecode, ip):
def local_read_byte():
b = bytecode[ip_ref[0]]
ip_ref[0] += 1
return b
- ip_ref = [ip] # to close over ip in Python 2 and 3
- n_state, n_exc_stack, scope_flags, n_pos_args, n_kwonly_args, n_def_pos_args = read_prelude_sig(local_read_byte)
+
+ ip_ref = [ip] # to close over ip in Python 2 and 3
+ (
+ n_state,
+ n_exc_stack,
+ scope_flags,
+ n_pos_args,
+ n_kwonly_args,
+ n_def_pos_args,
+ ) = read_prelude_sig(local_read_byte)
n_info, n_cell = read_prelude_size(local_read_byte)
ip = ip_ref[0]
@@ -196,19 +218,21 @@ def extract_prelude(bytecode, ip):
# ip2 points to simple_name qstr
return ip, ip2, (n_state, n_exc_stack, scope_flags, n_pos_args, n_kwonly_args, n_def_pos_args)
+
class MPFunTable:
pass
+
class RawCode(object):
# a set of all escaped names, to make sure they are unique
escaped_names = set()
# convert code kind number to string
code_kind_str = {
- MP_CODE_BYTECODE: 'MP_CODE_BYTECODE',
- MP_CODE_NATIVE_PY: 'MP_CODE_NATIVE_PY',
- MP_CODE_NATIVE_VIPER: 'MP_CODE_NATIVE_VIPER',
- MP_CODE_NATIVE_ASM: 'MP_CODE_NATIVE_ASM',
+ MP_CODE_BYTECODE: "MP_CODE_BYTECODE",
+ MP_CODE_NATIVE_PY: "MP_CODE_NATIVE_PY",
+ MP_CODE_NATIVE_VIPER: "MP_CODE_NATIVE_VIPER",
+ MP_CODE_NATIVE_ASM: "MP_CODE_NATIVE_ASM",
}
def __init__(self, code_kind, bytecode, prelude_offset, qstrs, objs, raw_codes):
@@ -237,7 +261,7 @@ class RawCode(object):
def dump(self):
# dump children first
for rc in self.raw_codes:
- rc.freeze('')
+ rc.freeze("")
# TODO
def freeze_children(self, parent_name):
@@ -252,32 +276,39 @@ class RawCode(object):
# emit children first
for rc in self.raw_codes:
- rc.freeze(self.escaped_name + '_')
+ rc.freeze(self.escaped_name + "_")
def freeze_constants(self):
# generate constant objects
for i, obj in enumerate(self.objs):
- obj_name = 'const_obj_%s_%u' % (self.escaped_name, i)
+ obj_name = "const_obj_%s_%u" % (self.escaped_name, i)
if obj is MPFunTable:
pass
elif obj is Ellipsis:
- print('#define %s mp_const_ellipsis_obj' % obj_name)
+ print("#define %s mp_const_ellipsis_obj" % obj_name)
elif is_str_type(obj) or is_bytes_type(obj):
if is_str_type(obj):
- obj = bytes_cons(obj, 'utf8')
- obj_type = 'mp_type_str'
+ obj = bytes_cons(obj, "utf8")
+ obj_type = "mp_type_str"
else:
- obj_type = 'mp_type_bytes'
- print('STATIC const mp_obj_str_t %s = {{&%s}, %u, %u, (const byte*)"%s"};'
- % (obj_name, obj_type, qstrutil.compute_hash(obj, config.MICROPY_QSTR_BYTES_IN_HASH),
- len(obj), ''.join(('\\x%02x' % b) for b in obj)))
+ obj_type = "mp_type_bytes"
+ print(
+ 'STATIC const mp_obj_str_t %s = {{&%s}, %u, %u, (const byte*)"%s"};'
+ % (
+ obj_name,
+ obj_type,
+ qstrutil.compute_hash(obj, config.MICROPY_QSTR_BYTES_IN_HASH),
+ len(obj),
+ "".join(("\\x%02x" % b) for b in obj),
+ )
+ )
elif is_int_type(obj):
if config.MICROPY_LONGINT_IMPL == config.MICROPY_LONGINT_IMPL_NONE:
# TODO check if we can actually fit this long-int into a small-int
- raise FreezeError(self, 'target does not support long int')
+ raise FreezeError(self, "target does not support long int")
elif config.MICROPY_LONGINT_IMPL == config.MICROPY_LONGINT_IMPL_LONGLONG:
# TODO
- raise FreezeError(self, 'freezing int to long-long is not implemented')
+ raise FreezeError(self, "freezing int to long-long is not implemented")
elif config.MICROPY_LONGINT_IMPL == config.MICROPY_LONGINT_IMPL_MPZ:
neg = 0
if obj < 0:
@@ -290,140 +321,175 @@ class RawCode(object):
digs.append(z & ((1 << bits_per_dig) - 1))
z >>= bits_per_dig
ndigs = len(digs)
- digs = ','.join(('%#x' % d) for d in digs)
- print('STATIC const mp_obj_int_t %s = {{&mp_type_int}, '
- '{.neg=%u, .fixed_dig=1, .alloc=%u, .len=%u, .dig=(uint%u_t*)(const uint%u_t[]){%s}}};'
- % (obj_name, neg, ndigs, ndigs, bits_per_dig, bits_per_dig, digs))
+ digs = ",".join(("%#x" % d) for d in digs)
+ print(
+ "STATIC const mp_obj_int_t %s = {{&mp_type_int}, "
+ "{.neg=%u, .fixed_dig=1, .alloc=%u, .len=%u, .dig=(uint%u_t*)(const uint%u_t[]){%s}}};"
+ % (obj_name, neg, ndigs, ndigs, bits_per_dig, bits_per_dig, digs)
+ )
elif type(obj) is float:
- print('#if MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_A || MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_B')
- print('STATIC const mp_obj_float_t %s = {{&mp_type_float}, %.16g};'
- % (obj_name, obj))
- print('#endif')
+ print(
+ "#if MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_A || MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_B"
+ )
+ print(
+ "STATIC const mp_obj_float_t %s = {{&mp_type_float}, %.16g};" % (obj_name, obj)
+ )
+ print("#endif")
elif type(obj) is complex:
- print('STATIC const mp_obj_complex_t %s = {{&mp_type_complex}, %.16g, %.16g};'
- % (obj_name, obj.real, obj.imag))
+ print(
+ "STATIC const mp_obj_complex_t %s = {{&mp_type_complex}, %.16g, %.16g};"
+ % (obj_name, obj.real, obj.imag)
+ )
else:
- raise FreezeError(self, 'freezing of object %r is not implemented' % (obj,))
+ raise FreezeError(self, "freezing of object %r is not implemented" % (obj,))
# generate constant table, if it has any entries
const_table_len = len(self.qstrs) + len(self.objs) + len(self.raw_codes)
if const_table_len:
- print('STATIC const mp_rom_obj_t const_table_data_%s[%u] = {'
- % (self.escaped_name, const_table_len))
+ print(
+ "STATIC const mp_rom_obj_t const_table_data_%s[%u] = {"
+ % (self.escaped_name, const_table_len)
+ )
for qst in self.qstrs:
- print(' MP_ROM_QSTR(%s),' % global_qstrs[qst].qstr_id)
+ print(" MP_ROM_QSTR(%s)," % global_qstrs[qst].qstr_id)
for i in range(len(self.objs)):
if self.objs[i] is MPFunTable:
- print(' &mp_fun_table,')
+ print(" &mp_fun_table,")
elif type(self.objs[i]) is float:
- print('#if MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_A || MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_B')
- print(' MP_ROM_PTR(&const_obj_%s_%u),' % (self.escaped_name, i))
- print('#elif MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_C')
- n = struct.unpack('<I', struct.pack('<f', self.objs[i]))[0]
+ print(
+ "#if MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_A || MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_B"
+ )
+ print(" MP_ROM_PTR(&const_obj_%s_%u)," % (self.escaped_name, i))
+ print("#elif MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_C")
+ n = struct.unpack("<I", struct.pack("<f", self.objs[i]))[0]
n = ((n & ~0x3) | 2) + 0x80800000
- print(' (mp_rom_obj_t)(0x%08x),' % (n,))
- print('#elif MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_D')
- n = struct.unpack('<Q', struct.pack('<d', self.objs[i]))[0]
+ print(" (mp_rom_obj_t)(0x%08x)," % (n,))
+ print("#elif MICROPY_OBJ_REPR == MICROPY_OBJ_REPR_D")
+ n = struct.unpack("<Q", struct.pack("<d", self.objs[i]))[0]
n += 0x8004000000000000
- print(' (mp_rom_obj_t)(0x%016x),' % (n,))
- print('#endif')
+ print(" (mp_rom_obj_t)(0x%016x)," % (n,))
+ print("#endif")
else:
- print(' MP_ROM_PTR(&const_obj_%s_%u),' % (self.escaped_name, i))
+ print(" MP_ROM_PTR(&const_obj_%s_%u)," % (self.escaped_name, i))
for rc in self.raw_codes:
- print(' MP_ROM_PTR(&raw_code_%s),' % rc.escaped_name)
- print('};')
+ print(" MP_ROM_PTR(&raw_code_%s)," % rc.escaped_name)
+ print("};")
def freeze_module(self, qstr_links=(), type_sig=0):
# generate module
- if self.simple_name.str != '<module>':
- print('STATIC ', end='')
- print('const mp_raw_code_t raw_code_%s = {' % self.escaped_name)
- print(' .kind = %s,' % RawCode.code_kind_str[self.code_kind])
- print(' .scope_flags = 0x%02x,' % self.prelude[2])
- print(' .n_pos_args = %u,' % self.prelude[3])
- print(' .fun_data = fun_data_%s,' % self.escaped_name)
+ if self.simple_name.str != "<module>":
+ print("STATIC ", end="")
+ print("const mp_raw_code_t raw_code_%s = {" % self.escaped_name)
+ print(" .kind = %s," % RawCode.code_kind_str[self.code_kind])
+ print(" .scope_flags = 0x%02x," % self.prelude[2])
+ print(" .n_pos_args = %u," % self.prelude[3])
+ print(" .fun_data = fun_data_%s," % self.escaped_name)
if len(self.qstrs) + len(self.objs) + len(self.raw_codes):
- print(' .const_table = (mp_uint_t*)const_table_data_%s,' % self.escaped_name)
+ print(" .const_table = (mp_uint_t*)const_table_data_%s," % self.escaped_name)
else:
- print(' .const_table = NULL,')
- print(' #if MICROPY_PERSISTENT_CODE_SAVE')
- print(' .fun_data_len = %u,' % len(self.bytecode))
- print(' .n_obj = %u,' % len(self.objs))
- print(' .n_raw_code = %u,' % len(self.raw_codes))
+ print(" .const_table = NULL,")
+ print(" #if MICROPY_PERSISTENT_CODE_SAVE")
+ print(" .fun_data_len = %u," % len(self.bytecode))
+ print(" .n_obj = %u," % len(self.objs))
+ print(" .n_raw_code = %u," % len(self.raw_codes))
if self.code_kind == MP_CODE_BYTECODE:
- print(' #if MICROPY_PY_SYS_SETTRACE')
- print(' .prelude = {')
- print(' .n_state = %u,' % self.prelude[0])
- print(' .n_exc_stack = %u,' % self.prelude[1])
- print(' .scope_flags = %u,' % self.prelude[2])
- print(' .n_pos_args = %u,' % self.prelude[3])
- print(' .n_kwonly_args = %u,' % self.prelude[4])
- print(' .n_def_pos_args = %u,' % self.prelude[5])
- print(' .qstr_block_name = %s,' % self.simple_name.qstr_id)
- print(' .qstr_source_file = %s,' % self.source_file.qstr_id)
- print(' .line_info = fun_data_%s + %u,' % (self.escaped_name, 0)) # TODO
- print(' .opcodes = fun_data_%s + %u,' % (self.escaped_name, self.ip))
- print(' },')
- print(' .line_of_definition = %u,' % 0) # TODO
- print(' #endif')
- print(' #if MICROPY_EMIT_MACHINE_CODE')
- print(' .prelude_offset = %u,' % self.prelude_offset)
- print(' .n_qstr = %u,' % len(qstr_links))
- print(' .qstr_link = NULL,') # TODO
- print(' #endif')
- print(' #endif')
- print(' #if MICROPY_EMIT_MACHINE_CODE')
- print(' .type_sig = %u,' % type_sig)
- print(' #endif')
- print('};')
+ print(" #if MICROPY_PY_SYS_SETTRACE")
+ print(" .prelude = {")
+ print(" .n_state = %u," % self.prelude[0])
+ print(" .n_exc_stack = %u," % self.prelude[1])
+ print(" .scope_flags = %u," % self.prelude[2])
+ print(" .n_pos_args = %u," % self.prelude[3])
+ print(" .n_kwonly_args = %u," % self.prelude[4])
+ print(" .n_def_pos_args = %u," % self.prelude[5])
+ print(" .qstr_block_name = %s," % self.simple_name.qstr_id)
+ print(" .qstr_source_file = %s," % self.source_file.qstr_id)
+ print(" .line_info = fun_data_%s + %u," % (self.escaped_name, 0)) # TODO
+ print(" .opcodes = fun_data_%s + %u," % (self.escaped_name, self.ip))
+ print(" },")
+ print(" .line_of_definition = %u," % 0) # TODO
+ print(" #endif")
+ print(" #if MICROPY_EMIT_MACHINE_CODE")
+ print(" .prelude_offset = %u," % self.prelude_offset)
+ print(" .n_qstr = %u," % len(qstr_links))
+ print(" .qstr_link = NULL,") # TODO
+ print(" #endif")
+ print(" #endif")
+ print(" #if MICROPY_EMIT_MACHINE_CODE")
+ print(" .type_sig = %u," % type_sig)
+ print(" #endif")
+ print("};")
+
class RawCodeBytecode(RawCode):
def __init__(self, bytecode, qstrs, objs, raw_codes):
- super(RawCodeBytecode, self).__init__(MP_CODE_BYTECODE, bytecode, 0, qstrs, objs, raw_codes)
+ super(RawCodeBytecode, self).__init__(
+ MP_CODE_BYTECODE, bytecode, 0, qstrs, objs, raw_codes
+ )
def freeze(self, parent_name):
self.freeze_children(parent_name)
# generate bytecode data
print()
- print('// frozen bytecode for file %s, scope %s%s' % (self.source_file.str, parent_name, self.simple_name.str))
- print('STATIC ', end='')
+ print(
+ "// frozen bytecode for file %s, scope %s%s"
+ % (self.source_file.str, parent_name, self.simple_name.str)
+ )
+ print("STATIC ", end="")
if not config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE:
- print('const ', end='')
- print('byte fun_data_%s[%u] = {' % (self.escaped_name, len(self.bytecode)))
- print(' ', end='')
+ print("const ", end="")
+ print("byte fun_data_%s[%u] = {" % (self.escaped_name, len(self.bytecode)))
+ print(" ", end="")
for i in range(self.ip2):
- print(' 0x%02x,' % self.bytecode[i], end='')
+ print(" 0x%02x," % self.bytecode[i], end="")
print()
- print(' ', self.simple_name.qstr_id, '& 0xff,', self.simple_name.qstr_id, '>> 8,')
- print(' ', self.source_file.qstr_id, '& 0xff,', self.source_file.qstr_id, '>> 8,')
- print(' ', end='')
+ print(" ", self.simple_name.qstr_id, "& 0xff,", self.simple_name.qstr_id, ">> 8,")
+ print(" ", self.source_file.qstr_id, "& 0xff,", self.source_file.qstr_id, ">> 8,")
+ print(" ", end="")
for i in range(self.ip2 + 4, self.ip):
- print(' 0x%02x,' % self.bytecode[i], end='')
+ print(" 0x%02x," % self.bytecode[i], end="")
print()
ip = self.ip
while ip < len(self.bytecode):
f, sz = mp_opcode_format(self.bytecode, ip, True)
if f == 1:
qst = self._unpack_qstr(ip + 1).qstr_id
- extra = '' if sz == 3 else ' 0x%02x,' % self.bytecode[ip + 3]
- print(' ', '0x%02x,' % self.bytecode[ip], qst, '& 0xff,', qst, '>> 8,', extra)
+ extra = "" if sz == 3 else " 0x%02x," % self.bytecode[ip + 3]
+ print(" ", "0x%02x," % self.bytecode[ip], qst, "& 0xff,", qst, ">> 8,", extra)
else:
- print(' ', ''.join('0x%02x, ' % self.bytecode[ip + i] for i in range(sz)))
+ print(" ", "".join("0x%02x, " % self.bytecode[ip + i] for i in range(sz)))
ip += sz
- print('};')
+ print("};")
self.freeze_constants()
self.freeze_module()
+
class RawCodeNative(RawCode):
- def __init__(self, code_kind, fun_data, prelude_offset, prelude, qstr_links, qstrs, objs, raw_codes, type_sig):
- super(RawCodeNative, self).__init__(code_kind, fun_data, prelude_offset, qstrs, objs, raw_codes)
+ def __init__(
+ self,
+ code_kind,
+ fun_data,
+ prelude_offset,
+ prelude,
+ qstr_links,
+ qstrs,
+ objs,
+ raw_codes,
+ type_sig,
+ ):
+ super(RawCodeNative, self).__init__(
+ code_kind, fun_data, prelude_offset, qstrs, objs, raw_codes
+ )
self.prelude = prelude
self.qstr_links = qstr_links
self.type_sig = type_sig
- if config.native_arch in (MP_NATIVE_ARCH_X86, MP_NATIVE_ARCH_X64,
- MP_NATIVE_ARCH_XTENSA, MP_NATIVE_ARCH_XTENSAWIN):
+ if config.native_arch in (
+ MP_NATIVE_ARCH_X86,
+ MP_NATIVE_ARCH_X64,
+ MP_NATIVE_ARCH_XTENSA,
+ MP_NATIVE_ARCH_XTENSAWIN,
+ ):
self.fun_data_attributes = '__attribute__((section(".text,\\"ax\\",@progbits # ")))'
else:
self.fun_data_attributes = '__attribute__((section(".text,\\"ax\\",%progbits @ ")))'
@@ -431,40 +497,49 @@ class RawCodeNative(RawCode):
# Allow single-byte alignment by default for x86/x64.
# ARM needs word alignment, ARM Thumb needs halfword, due to instruction size.
# Xtensa needs word alignment due to the 32-bit constant table embedded in the code.
- if config.native_arch in (MP_NATIVE_ARCH_ARMV6, MP_NATIVE_ARCH_XTENSA, MP_NATIVE_ARCH_XTENSAWIN):
+ if config.native_arch in (
+ MP_NATIVE_ARCH_ARMV6,
+ MP_NATIVE_ARCH_XTENSA,
+ MP_NATIVE_ARCH_XTENSAWIN,
+ ):
# ARMV6 or Xtensa -- four byte align.
- self.fun_data_attributes += ' __attribute__ ((aligned (4)))'
+ self.fun_data_attributes += " __attribute__ ((aligned (4)))"
elif MP_NATIVE_ARCH_ARMV6M <= config.native_arch <= MP_NATIVE_ARCH_ARMV7EMDP:
# ARMVxxM -- two byte align.
- self.fun_data_attributes += ' __attribute__ ((aligned (2)))'
+ self.fun_data_attributes += " __attribute__ ((aligned (2)))"
def _asm_thumb_rewrite_mov(self, pc, val):
- print(' (%u & 0xf0) | (%s >> 12),' % (self.bytecode[pc], val), end='')
- print(' (%u & 0xfb) | (%s >> 9 & 0x04),' % (self.bytecode[pc + 1], val), end='')
- print(' (%s & 0xff),' % (val,), end='')
- print(' (%u & 0x07) | (%s >> 4 & 0x70),' % (self.bytecode[pc + 3], val))
+ print(" (%u & 0xf0) | (%s >> 12)," % (self.bytecode[pc], val), end="")
+ print(" (%u & 0xfb) | (%s >> 9 & 0x04)," % (self.bytecode[pc + 1], val), end="")
+ print(" (%s & 0xff)," % (val,), end="")
+ print(" (%u & 0x07) | (%s >> 4 & 0x70)," % (self.bytecode[pc + 3], val))
def _link_qstr(self, pc, kind, qst):
if kind == 0:
# Generic 16-bit link
- print(' %s & 0xff, %s >> 8,' % (qst, qst))
+ print(" %s & 0xff, %s >> 8," % (qst, qst))
return 2
else:
# Architecture-specific link
is_obj = kind == 2
if is_obj:
- qst = '((uintptr_t)MP_OBJ_NEW_QSTR(%s))' % qst
+ qst = "((uintptr_t)MP_OBJ_NEW_QSTR(%s))" % qst
if config.native_arch in (
- MP_NATIVE_ARCH_X86, MP_NATIVE_ARCH_X64,
- MP_NATIVE_ARCH_XTENSA, MP_NATIVE_ARCH_XTENSAWIN
- ):
- print(' %s & 0xff, (%s >> 8) & 0xff, (%s >> 16) & 0xff, %s >> 24,' % (qst, qst, qst, qst))
+ MP_NATIVE_ARCH_X86,
+ MP_NATIVE_ARCH_X64,
+ MP_NATIVE_ARCH_XTENSA,
+ MP_NATIVE_ARCH_XTENSAWIN,
+ ):
+ print(
+ " %s & 0xff, (%s >> 8) & 0xff, (%s >> 16) & 0xff, %s >> 24,"
+ % (qst, qst, qst, qst)
+ )
return 4
elif MP_NATIVE_ARCH_ARMV6M <= config.native_arch <= MP_NATIVE_ARCH_ARMV7EMDP:
if is_obj:
# qstr object, movw and movt
self._asm_thumb_rewrite_mov(pc, qst)
- self._asm_thumb_rewrite_mov(pc + 4, '(%s >> 16)' % qst)
+ self._asm_thumb_rewrite_mov(pc + 4, "(%s >> 16)" % qst)
return 8
else:
# qstr number, movw instruction
@@ -474,20 +549,26 @@ class RawCodeNative(RawCode):
assert 0
def freeze(self, parent_name):
- if self.prelude[2] & ~0x0f:
- raise FreezeError('unable to freeze code with relocations')
+ if self.prelude[2] & ~0x0F:
+ raise FreezeError("unable to freeze code with relocations")
self.freeze_children(parent_name)
# generate native code data
print()
if self.code_kind == MP_CODE_NATIVE_PY:
- print('// frozen native code for file %s, scope %s%s' % (self.source_file.str, parent_name, self.simple_name.str))
+ print(
+ "// frozen native code for file %s, scope %s%s"
+ % (self.source_file.str, parent_name, self.simple_name.str)
+ )
elif self.code_kind == MP_CODE_NATIVE_VIPER:
- print('// frozen viper code for scope %s' % (parent_name,))
+ print("// frozen viper code for scope %s" % (parent_name,))
else:
- print('// frozen assembler code for scope %s' % (parent_name,))
- print('STATIC const byte fun_data_%s[%u] %s = {' % (self.escaped_name, len(self.bytecode), self.fun_data_attributes))
+ print("// frozen assembler code for scope %s" % (parent_name,))
+ print(
+ "STATIC const byte fun_data_%s[%u] %s = {"
+ % (self.escaped_name, len(self.bytecode), self.fun_data_attributes)
+ )
if self.code_kind == MP_CODE_NATIVE_PY:
i_top = self.prelude_offset
@@ -507,31 +588,32 @@ class RawCodeNative(RawCode):
i16 = min(i + 16, i_top)
if qi < len(self.qstr_links):
i16 = min(i16, self.qstr_links[qi][0])
- print(' ', end='')
+ print(" ", end="")
for ii in range(i, i16):
- print(' 0x%02x,' % self.bytecode[ii], end='')
+ print(" 0x%02x," % self.bytecode[ii], end="")
print()
i = i16
if self.code_kind == MP_CODE_NATIVE_PY:
- print(' ', end='')
+ print(" ", end="")
for i in range(self.prelude_offset, self.ip2):
- print(' 0x%02x,' % self.bytecode[i], end='')
+ print(" 0x%02x," % self.bytecode[i], end="")
print()
- print(' ', self.simple_name.qstr_id, '& 0xff,', self.simple_name.qstr_id, '>> 8,')
- print(' ', self.source_file.qstr_id, '& 0xff,', self.source_file.qstr_id, '>> 8,')
+ print(" ", self.simple_name.qstr_id, "& 0xff,", self.simple_name.qstr_id, ">> 8,")
+ print(" ", self.source_file.qstr_id, "& 0xff,", self.source_file.qstr_id, ">> 8,")
- print(' ', end='')
+ print(" ", end="")
for i in range(self.ip2 + 4, self.ip):
- print(' 0x%02x,' % self.bytecode[i], end='')
+ print(" 0x%02x," % self.bytecode[i], end="")
print()
- print('};')
+ print("};")
self.freeze_constants()
self.freeze_module(self.qstr_links, self.type_sig)
+
class BytecodeBuffer:
def __init__(self, size):
self.buf = bytearray(size)
@@ -544,21 +626,24 @@ class BytecodeBuffer:
self.buf[self.idx] = b
self.idx += 1
+
def read_byte(f, out=None):
b = bytes_cons(f.read(1))[0]
if out is not None:
out.append(b)
return b
+
def read_uint(f, out=None):
i = 0
while True:
b = read_byte(f, out)
- i = (i << 7) | (b & 0x7f)
+ i = (i << 7) | (b & 0x7F)
if b & 0x80 == 0:
break
return i
+
def read_qstr(f, qstr_win):
ln = read_uint(f)
if ln == 0:
@@ -568,44 +653,55 @@ def read_qstr(f, qstr_win):
# qstr in table
return qstr_win.access(ln >> 1)
ln >>= 1
- data = str_cons(f.read(ln), 'utf8')
+ data = str_cons(f.read(ln), "utf8")
global_qstrs.append(QStrType(data))
qstr_win.push(len(global_qstrs) - 1)
return len(global_qstrs) - 1
+
def read_obj(f):
obj_type = f.read(1)
- if obj_type == b'e':
+ if obj_type == b"e":
return Ellipsis
else:
buf = f.read(read_uint(f))
- if obj_type == b's':
- return str_cons(buf, 'utf8')
- elif obj_type == b'b':
+ if obj_type == b"s":
+ return str_cons(buf, "utf8")
+ elif obj_type == b"b":
return bytes_cons(buf)
- elif obj_type == b'i':
- return int(str_cons(buf, 'ascii'), 10)
- elif obj_type == b'f':
- return float(str_cons(buf, 'ascii'))
- elif obj_type == b'c':
- return complex(str_cons(buf, 'ascii'))
+ elif obj_type == b"i":
+ return int(str_cons(buf, "ascii"), 10)
+ elif obj_type == b"f":
+ return float(str_cons(buf, "ascii"))
+ elif obj_type == b"c":
+ return complex(str_cons(buf, "ascii"))
else:
assert 0
+
def read_prelude(f, bytecode, qstr_win):
- n_state, n_exc_stack, scope_flags, n_pos_args, n_kwonly_args, n_def_pos_args = read_prelude_sig(lambda: read_byte(f, bytecode))
+ (
+ n_state,
+ n_exc_stack,
+ scope_flags,
+ n_pos_args,
+ n_kwonly_args,
+ n_def_pos_args,
+ ) = read_prelude_sig(lambda: read_byte(f, bytecode))
n_info, n_cell = read_prelude_size(lambda: read_byte(f, bytecode))
- read_qstr_and_pack(f, bytecode, qstr_win) # simple_name
- read_qstr_and_pack(f, bytecode, qstr_win) # source_file
+ read_qstr_and_pack(f, bytecode, qstr_win) # simple_name
+ read_qstr_and_pack(f, bytecode, qstr_win) # source_file
for _ in range(n_info - 4 + n_cell):
read_byte(f, bytecode)
return n_state, n_exc_stack, scope_flags, n_pos_args, n_kwonly_args, n_def_pos_args
+
def read_qstr_and_pack(f, bytecode, qstr_win):
qst = read_qstr(f, qstr_win)
- bytecode.append(qst & 0xff)
+ bytecode.append(qst & 0xFF)
bytecode.append(qst >> 8)
+
def read_bytecode(file, bytecode, qstr_win):
while not bytecode.is_full():
op = read_byte(file, bytecode)
@@ -620,6 +716,7 @@ def read_bytecode(file, bytecode, qstr_win):
for _ in range(sz):
read_byte(file, bytecode)
+
def read_raw_code(f, qstr_win):
kind_len = read_uint(f)
kind = (kind_len & 3) + MP_CODE_BYTECODE
@@ -645,9 +742,9 @@ def read_raw_code(f, qstr_win):
if kind == MP_CODE_NATIVE_PY:
prelude_offset = read_uint(f)
_, name_idx, prelude = extract_prelude(fun_data.buf, prelude_offset)
- fun_data.idx = name_idx # rewind to where qstrs are in prelude
- read_qstr_and_pack(f, fun_data, qstr_win) # simple_name
- read_qstr_and_pack(f, fun_data, qstr_win) # source_file
+ fun_data.idx = name_idx # rewind to where qstrs are in prelude
+ read_qstr_and_pack(f, fun_data, qstr_win) # simple_name
+ read_qstr_and_pack(f, fun_data, qstr_win) # source_file
else:
prelude_offset = None
scope_flags = read_uint(f)
@@ -673,15 +770,26 @@ def read_raw_code(f, qstr_win):
if kind == MP_CODE_BYTECODE:
return RawCodeBytecode(fun_data.buf, qstrs, objs, raw_codes)
else:
- return RawCodeNative(kind, fun_data.buf, prelude_offset, prelude, qstr_links, qstrs, objs, raw_codes, type_sig)
+ return RawCodeNative(
+ kind,
+ fun_data.buf,
+ prelude_offset,
+ prelude,
+ qstr_links,
+ qstrs,
+ objs,
+ raw_codes,
+ type_sig,
+ )
+
def read_mpy(filename):
- with open(filename, 'rb') as f:
+ with open(filename, "rb") as f:
header = bytes_cons(f.read(4))
- if header[0] != ord('M'):
- raise Exception('not a valid .mpy file')
+ if header[0] != ord("M"):
+ raise Exception("not a valid .mpy file")
if header[1] != config.MPY_VERSION:
- raise Exception('incompatible .mpy version')
+ raise Exception("incompatible .mpy version")
feature_byte = header[2]
qw_size = read_uint(f)
config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE = (feature_byte & 1) != 0
@@ -691,7 +799,7 @@ def read_mpy(filename):
if config.native_arch == MP_NATIVE_ARCH_NONE:
config.native_arch = mpy_native_arch
elif config.native_arch != mpy_native_arch:
- raise Exception('native architecture mismatch')
+ raise Exception("native architecture mismatch")
config.mp_small_int_bits = header[3]
qstr_win = QStrWindow(qw_size)
rc = read_raw_code(f, qstr_win)
@@ -699,10 +807,12 @@ def read_mpy(filename):
rc.qstr_win_size = qw_size
return rc
+
def dump_mpy(raw_codes):
for rc in raw_codes:
rc.dump()
+
def freeze_mpy(base_qstrs, raw_codes):
# add to qstrs
new = {}
@@ -720,156 +830,172 @@ def freeze_mpy(base_qstrs, raw_codes):
print('#include "py/nativeglue.h"')
print()
- print('#if MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE != %u' % config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE)
+ print(
+ "#if MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE != %u"
+ % config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE
+ )
print('#error "incompatible MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE"')
- print('#endif')
+ print("#endif")
print()
- print('#if MICROPY_LONGINT_IMPL != %u' % config.MICROPY_LONGINT_IMPL)
+ print("#if MICROPY_LONGINT_IMPL != %u" % config.MICROPY_LONGINT_IMPL)
print('#error "incompatible MICROPY_LONGINT_IMPL"')
- print('#endif')
+ print("#endif")
print()
if config.MICROPY_LONGINT_IMPL == config.MICROPY_LONGINT_IMPL_MPZ:
- print('#if MPZ_DIG_SIZE != %u' % config.MPZ_DIG_SIZE)
+ print("#if MPZ_DIG_SIZE != %u" % config.MPZ_DIG_SIZE)
print('#error "incompatible MPZ_DIG_SIZE"')
- print('#endif')
+ print("#endif")
print()
-
- print('#if MICROPY_PY_BUILTINS_FLOAT')
- print('typedef struct _mp_obj_float_t {')
- print(' mp_obj_base_t base;')
- print(' mp_float_t value;')
- print('} mp_obj_float_t;')
- print('#endif')
+ print("#if MICROPY_PY_BUILTINS_FLOAT")
+ print("typedef struct _mp_obj_float_t {")
+ print(" mp_obj_base_t base;")
+ print(" mp_float_t value;")
+ print("} mp_obj_float_t;")
+ print("#endif")
print()
- print('#if MICROPY_PY_BUILTINS_COMPLEX')
- print('typedef struct _mp_obj_complex_t {')
- print(' mp_obj_base_t base;')
- print(' mp_float_t real;')
- print(' mp_float_t imag;')
- print('} mp_obj_complex_t;')
- print('#endif')
+ print("#if MICROPY_PY_BUILTINS_COMPLEX")
+ print("typedef struct _mp_obj_complex_t {")
+ print(" mp_obj_base_t base;")
+ print(" mp_float_t real;")
+ print(" mp_float_t imag;")
+ print("} mp_obj_complex_t;")
+ print("#endif")
print()
if len(new) > 0:
- print('enum {')
+ print("enum {")
for i in range(len(new)):
if i == 0:
- print(' MP_QSTR_%s = MP_QSTRnumber_of,' % new[i][1])
+ print(" MP_QSTR_%s = MP_QSTRnumber_of," % new[i][1])
else:
- print(' MP_QSTR_%s,' % new[i][1])
- print('};')
+ print(" MP_QSTR_%s," % new[i][1])
+ print("};")
# As in qstr.c, set so that the first dynamically allocated pool is twice this size; must be <= the len
qstr_pool_alloc = min(len(new), 10)
print()
- print('extern const qstr_pool_t mp_qstr_const_pool;');
- print('const qstr_pool_t mp_qstr_frozen_const_pool = {')
- print(' (qstr_pool_t*)&mp_qstr_const_pool, // previous pool')
- print(' MP_QSTRnumber_of, // previous pool size')
- print(' %u, // allocated entries' % qstr_pool_alloc)
- print(' %u, // used entries' % len(new))
- print(' {')
+ print("extern const qstr_pool_t mp_qstr_const_pool;")
+ print("const qstr_pool_t mp_qstr_frozen_const_pool = {")
+ print(" (qstr_pool_t*)&mp_qstr_const_pool, // previous pool")
+ print(" MP_QSTRnumber_of, // previous pool size")
+ print(" %u, // allocated entries" % qstr_pool_alloc)
+ print(" %u, // used entries" % len(new))
+ print(" {")
for _, _, qstr in new:
- print(' %s,'
- % qstrutil.make_bytes(config.MICROPY_QSTR_BYTES_IN_LEN, config.MICROPY_QSTR_BYTES_IN_HASH, qstr))
- print(' },')
- print('};')
+ print(
+ " %s,"
+ % qstrutil.make_bytes(
+ config.MICROPY_QSTR_BYTES_IN_LEN, config.MICROPY_QSTR_BYTES_IN_HASH, qstr
+ )
+ )
+ print(" },")
+ print("};")
for rc in raw_codes:
- rc.freeze(rc.source_file.str.replace('/', '_')[:-3] + '_')
+ rc.freeze(rc.source_file.str.replace("/", "_")[:-3] + "_")
print()
- print('const char mp_frozen_mpy_names[] = {')
+ print("const char mp_frozen_mpy_names[] = {")
for rc in raw_codes:
module_name = rc.source_file.str
print('"%s\\0"' % module_name)
print('"\\0"};')
- print('const mp_raw_code_t *const mp_frozen_mpy_content[] = {')
+ print("const mp_raw_code_t *const mp_frozen_mpy_content[] = {")
for rc in raw_codes:
- print(' &raw_code_%s,' % rc.escaped_name)
- print('};')
+ print(" &raw_code_%s," % rc.escaped_name)
+ print("};")
+
def merge_mpy(raw_codes, output_file):
- assert len(raw_codes) <= 31 # so var-uints all fit in 1 byte
+ assert len(raw_codes) <= 31 # so var-uints all fit in 1 byte
merged_mpy = bytearray()
if len(raw_codes) == 1:
- with open(raw_codes[0].mpy_source_file, 'rb') as f:
+ with open(raw_codes[0].mpy_source_file, "rb") as f:
merged_mpy.extend(f.read())
else:
header = bytearray(5)
- header[0] = ord('M')
+ header[0] = ord("M")
header[1] = config.MPY_VERSION
- header[2] = (config.native_arch << 2
+ header[2] = (
+ config.native_arch << 2
| config.MICROPY_PY_BUILTINS_STR_UNICODE << 1
- | config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE)
+ | config.MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE
+ )
header[3] = config.mp_small_int_bits
- header[4] = 32 # qstr_win_size
+ header[4] = 32 # qstr_win_size
merged_mpy.extend(header)
bytecode = bytearray()
bytecode_len = 6 + len(raw_codes) * 4 + 2
- bytecode.append(bytecode_len << 2) # kind and length
- bytecode.append(0b00000000) # signature prelude
- bytecode.append(0b00001000) # size prelude
- bytecode.extend(b'\x00\x01') # MP_QSTR_
- bytecode.extend(b'\x00\x01') # MP_QSTR_
+ bytecode.append(bytecode_len << 2) # kind and length
+ bytecode.append(0b00000000) # signature prelude
+ bytecode.append(0b00001000) # size prelude
+ bytecode.extend(b"\x00\x01") # MP_QSTR_
+ bytecode.extend(b"\x00\x01") # MP_QSTR_
for idx in range(len(raw_codes)):
- bytecode.append(0x32) # MP_BC_MAKE_FUNCTION
- bytecode.append(idx) # index raw code
- bytecode.extend(b'\x34\x00') # MP_BC_CALL_FUNCTION, 0 args
- bytecode.extend(b'\x51\x63') # MP_BC_LOAD_NONE, MP_BC_RETURN_VALUE
+ bytecode.append(0x32) # MP_BC_MAKE_FUNCTION
+ bytecode.append(idx) # index raw code
+ bytecode.extend(b"\x34\x00") # MP_BC_CALL_FUNCTION, 0 args
+ bytecode.extend(b"\x51\x63") # MP_BC_LOAD_NONE, MP_BC_RETURN_VALUE
- bytecode.append(0) # n_obj
- bytecode.append(len(raw_codes)) # n_raw_code
+ bytecode.append(0) # n_obj
+ bytecode.append(len(raw_codes)) # n_raw_code
merged_mpy.extend(bytecode)
for rc in raw_codes:
- with open(rc.mpy_source_file, 'rb') as f:
- f.read(4) # skip header
- read_uint(f) # skip qstr_win_size
- data = f.read() # read rest of mpy file
+ with open(rc.mpy_source_file, "rb") as f:
+ f.read(4) # skip header
+ read_uint(f) # skip qstr_win_size
+ data = f.read() # read rest of mpy file
merged_mpy.extend(data)
if output_file is None:
sys.stdout.buffer.write(merged_mpy)
else:
- with open(output_file, 'wb') as f:
+ with open(output_file, "wb") as f:
f.write(merged_mpy)
+
def main():
import argparse
- cmd_parser = argparse.ArgumentParser(description='A tool to work with MicroPython .mpy files.')
- cmd_parser.add_argument('-d', '--dump', action='store_true',
- help='dump contents of files')
- cmd_parser.add_argument('-f', '--freeze', action='store_true',
- help='freeze files')
- cmd_parser.add_argument('--merge', action='store_true',
- help='merge multiple .mpy files into one')
- cmd_parser.add_argument('-q', '--qstr-header',
- help='qstr header file to freeze against')
- cmd_parser.add_argument('-mlongint-impl', choices=['none', 'longlong', 'mpz'], default='mpz',
- help='long-int implementation used by target (default mpz)')
- cmd_parser.add_argument('-mmpz-dig-size', metavar='N', type=int, default=16,
- help='mpz digit size used by target (default 16)')
- cmd_parser.add_argument('-o', '--output', default=None,
- help='output file')
- cmd_parser.add_argument('files', nargs='+',
- help='input .mpy files')
+
+ cmd_parser = argparse.ArgumentParser(description="A tool to work with MicroPython .mpy files.")
+ cmd_parser.add_argument("-d", "--dump", action="store_true", help="dump contents of files")
+ cmd_parser.add_argument("-f", "--freeze", action="store_true", help="freeze files")
+ cmd_parser.add_argument(
+ "--merge", action="store_true", help="merge multiple .mpy files into one"
+ )
+ cmd_parser.add_argument("-q", "--qstr-header", help="qstr header file to freeze against")
+ cmd_parser.add_argument(
+ "-mlongint-impl",
+ choices=["none", "longlong", "mpz"],
+ default="mpz",
+ help="long-int implementation used by target (default mpz)",
+ )
+ cmd_parser.add_argument(
+ "-mmpz-dig-size",
+ metavar="N",
+ type=int,
+ default=16,
+ help="mpz digit size used by target (default 16)",
+ )
+ cmd_parser.add_argument("-o", "--output", default=None, help="output file")
+ cmd_parser.add_argument("files", nargs="+", help="input .mpy files")
args = cmd_parser.parse_args()
# set config values relevant to target machine
config.MICROPY_LONGINT_IMPL = {
- 'none':config.MICROPY_LONGINT_IMPL_NONE,
- 'longlong':config.MICROPY_LONGINT_IMPL_LONGLONG,
- 'mpz':config.MICROPY_LONGINT_IMPL_MPZ,
+ "none": config.MICROPY_LONGINT_IMPL_NONE,
+ "longlong": config.MICROPY_LONGINT_IMPL_LONGLONG,
+ "mpz": config.MICROPY_LONGINT_IMPL_MPZ,
}[args.mlongint_impl]
config.MPZ_DIG_SIZE = args.mmpz_dig_size
config.native_arch = MP_NATIVE_ARCH_NONE
@@ -877,8 +1003,8 @@ def main():
# set config values for qstrs, and get the existing base set of qstrs
if args.qstr_header:
qcfgs, base_qstrs = qstrutil.parse_input_headers([args.qstr_header])
- config.MICROPY_QSTR_BYTES_IN_LEN = int(qcfgs['BYTES_IN_LEN'])
- config.MICROPY_QSTR_BYTES_IN_HASH = int(qcfgs['BYTES_IN_HASH'])
+ config.MICROPY_QSTR_BYTES_IN_LEN = int(qcfgs["BYTES_IN_LEN"])
+ config.MICROPY_QSTR_BYTES_IN_HASH = int(qcfgs["BYTES_IN_HASH"])
else:
config.MICROPY_QSTR_BYTES_IN_LEN = 1
config.MICROPY_QSTR_BYTES_IN_HASH = 1
@@ -897,5 +1023,6 @@ def main():
elif args.merge:
merged_mpy = merge_mpy(raw_codes, args.output)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
diff --git a/tools/mpy_cross_all.py b/tools/mpy_cross_all.py
index 2bda71e9b..d542bde42 100755
--- a/tools/mpy_cross_all.py
+++ b/tools/mpy_cross_all.py
@@ -6,7 +6,9 @@ import os.path
argparser = argparse.ArgumentParser(description="Compile all .py files to .mpy recursively")
argparser.add_argument("-o", "--out", help="output directory (default: input dir)")
argparser.add_argument("--target", help="select MicroPython target config")
-argparser.add_argument("-mcache-lookup-bc", action="store_true", help="cache map lookups in the bytecode")
+argparser.add_argument(
+ "-mcache-lookup-bc", action="store_true", help="cache map lookups in the bytecode"
+)
argparser.add_argument("dir", help="input directory")
args = argparser.parse_args()
@@ -26,13 +28,17 @@ for path, subdirs, files in os.walk(args.dir):
for f in files:
if f.endswith(".py"):
fpath = path + "/" + f
- #print(fpath)
+ # print(fpath)
out_fpath = args.out + "/" + fpath[path_prefix_len:-3] + ".mpy"
out_dir = os.path.dirname(out_fpath)
if not os.path.isdir(out_dir):
os.makedirs(out_dir)
- cmd = "mpy-cross -v -v %s -s %s %s -o %s" % (TARGET_OPTS.get(args.target, ""),
- fpath[path_prefix_len:], fpath, out_fpath)
- #print(cmd)
+ cmd = "mpy-cross -v -v %s -s %s %s -o %s" % (
+ TARGET_OPTS.get(args.target, ""),
+ fpath[path_prefix_len:],
+ fpath,
+ out_fpath,
+ )
+ # print(cmd)
res = os.system(cmd)
assert res == 0
diff --git a/tools/mpy_ld.py b/tools/mpy_ld.py
index 31c391299..32bc176cb 100755
--- a/tools/mpy_ld.py
+++ b/tools/mpy_ld.py
@@ -31,7 +31,7 @@ Link .o files to .mpy
import sys, os, struct, re
from elftools.elf import elffile
-sys.path.append(os.path.dirname(__file__) + '/../py')
+sys.path.append(os.path.dirname(__file__) + "/../py")
import makeqstrdata as qstrutil
# MicroPython constants
@@ -70,8 +70,8 @@ R_386_GOTPC = 10
R_ARM_THM_CALL = 10
R_XTENSA_DIFF32 = 19
R_XTENSA_SLOT0_OP = 20
-R_ARM_BASE_PREL = 25 # aka R_ARM_GOTPC
-R_ARM_GOT_BREL = 26 # aka R_ARM_GOT32
+R_ARM_BASE_PREL = 25 # aka R_ARM_GOTPC
+R_ARM_GOT_BREL = 26 # aka R_ARM_GOT32
R_ARM_THM_JUMP24 = 30
R_X86_64_REX_GOTPCRELX = 42
R_386_GOT32X = 43
@@ -79,25 +79,29 @@ R_386_GOT32X = 43
################################################################################
# Architecture configuration
+
def asm_jump_x86(entry):
- return struct.pack('<BI', 0xe9, entry - 5)
+ return struct.pack("<BI", 0xE9, entry - 5)
+
def asm_jump_arm(entry):
b_off = entry - 4
if b_off >> 11 == 0 or b_off >> 11 == -1:
# Signed value fits in 12 bits
- b0 = 0xe000 | (b_off >> 1 & 0x07ff)
+ b0 = 0xE000 | (b_off >> 1 & 0x07FF)
b1 = 0
else:
# Use large jump
- b0 = 0xf000 | (b_off >> 12 & 0x07ff)
- b1 = 0xb800 | (b_off >> 1 & 0x7ff)
- return struct.pack('<HH', b0, b1)
+ b0 = 0xF000 | (b_off >> 12 & 0x07FF)
+ b1 = 0xB800 | (b_off >> 1 & 0x7FF)
+ return struct.pack("<HH", b0, b1)
+
def asm_jump_xtensa(entry):
jump_offset = entry - 4
jump_op = jump_offset << 6 | 6
- return struct.pack('<BH', jump_op & 0xff, jump_op >> 8)
+ return struct.pack("<BH", jump_op & 0xFF, jump_op >> 8)
+
class ArchData:
def __init__(self, name, mpy_feature, qstr_entry_size, word_size, arch_got, asm_jump):
@@ -107,83 +111,117 @@ class ArchData:
self.word_size = word_size
self.arch_got = arch_got
self.asm_jump = asm_jump
- self.separate_rodata = name == 'EM_XTENSA' and qstr_entry_size == 4
+ self.separate_rodata = name == "EM_XTENSA" and qstr_entry_size == 4
+
ARCH_DATA = {
- 'x86': ArchData(
- 'EM_386',
- MP_NATIVE_ARCH_X86 << 2 | MICROPY_PY_BUILTINS_STR_UNICODE | MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE,
- 2, 4, (R_386_PC32, R_386_GOT32, R_386_GOT32X), asm_jump_x86,
+ "x86": ArchData(
+ "EM_386",
+ MP_NATIVE_ARCH_X86 << 2
+ | MICROPY_PY_BUILTINS_STR_UNICODE
+ | MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE,
+ 2,
+ 4,
+ (R_386_PC32, R_386_GOT32, R_386_GOT32X),
+ asm_jump_x86,
),
- 'x64': ArchData(
- 'EM_X86_64',
- MP_NATIVE_ARCH_X64 << 2 | MICROPY_PY_BUILTINS_STR_UNICODE | MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE,
- 2, 8, (R_X86_64_REX_GOTPCRELX,), asm_jump_x86,
+ "x64": ArchData(
+ "EM_X86_64",
+ MP_NATIVE_ARCH_X64 << 2
+ | MICROPY_PY_BUILTINS_STR_UNICODE
+ | MICROPY_OPT_CACHE_MAP_LOOKUP_IN_BYTECODE,
+ 2,
+ 8,
+ (R_X86_64_REX_GOTPCRELX,),
+ asm_jump_x86,
),
- 'armv7m': ArchData(
- 'EM_ARM',
+ "armv7m": ArchData(
+ "EM_ARM",
MP_NATIVE_ARCH_ARMV7M << 2 | MICROPY_PY_BUILTINS_STR_UNICODE,
- 2, 4, (R_ARM_GOT_BREL,), asm_jump_arm,
+ 2,
+ 4,
+ (R_ARM_GOT_BREL,),
+ asm_jump_arm,
),
- 'armv7emsp': ArchData(
- 'EM_ARM',
+ "armv7emsp": ArchData(
+ "EM_ARM",
MP_NATIVE_ARCH_ARMV7EMSP << 2 | MICROPY_PY_BUILTINS_STR_UNICODE,
- 2, 4, (R_ARM_GOT_BREL,), asm_jump_arm,
+ 2,
+ 4,
+ (R_ARM_GOT_BREL,),
+ asm_jump_arm,
),
- 'armv7emdp': ArchData(
- 'EM_ARM',
+ "armv7emdp": ArchData(
+ "EM_ARM",
MP_NATIVE_ARCH_ARMV7EMDP << 2 | MICROPY_PY_BUILTINS_STR_UNICODE,
- 2, 4, (R_ARM_GOT_BREL,), asm_jump_arm,
+ 2,
+ 4,
+ (R_ARM_GOT_BREL,),
+ asm_jump_arm,
),
- 'xtensa': ArchData(
- 'EM_XTENSA',
+ "xtensa": ArchData(
+ "EM_XTENSA",
MP_NATIVE_ARCH_XTENSA << 2 | MICROPY_PY_BUILTINS_STR_UNICODE,
- 2, 4, (R_XTENSA_32, R_XTENSA_PLT), asm_jump_xtensa,
+ 2,
+ 4,
+ (R_XTENSA_32, R_XTENSA_PLT),
+ asm_jump_xtensa,
),
- 'xtensawin': ArchData(
- 'EM_XTENSA',
+ "xtensawin": ArchData(
+ "EM_XTENSA",
MP_NATIVE_ARCH_XTENSAWIN << 2 | MICROPY_PY_BUILTINS_STR_UNICODE,
- 4, 4, (R_XTENSA_32, R_XTENSA_PLT), asm_jump_xtensa,
+ 4,
+ 4,
+ (R_XTENSA_32, R_XTENSA_PLT),
+ asm_jump_xtensa,
),
}
################################################################################
# Helper functions
+
def align_to(value, align):
return (value + align - 1) & ~(align - 1)
+
def unpack_u24le(data, offset):
return data[offset] | data[offset + 1] << 8 | data[offset + 2] << 16
+
def pack_u24le(data, offset, value):
- data[offset] = value & 0xff
- data[offset + 1] = value >> 8 & 0xff
- data[offset + 2] = value >> 16 & 0xff
+ data[offset] = value & 0xFF
+ data[offset + 1] = value >> 8 & 0xFF
+ data[offset + 2] = value >> 16 & 0xFF
+
def xxd(text):
for i in range(0, len(text), 16):
- print('{:08x}:'.format(i), end='')
+ print("{:08x}:".format(i), end="")
for j in range(4):
off = i + j * 4
if off < len(text):
- d = int.from_bytes(text[off:off + 4], 'little')
- print(' {:08x}'.format(d), end='')
+ d = int.from_bytes(text[off : off + 4], "little")
+ print(" {:08x}".format(d), end="")
print()
+
# Smaller numbers are enabled first
LOG_LEVEL_1 = 1
LOG_LEVEL_2 = 2
LOG_LEVEL_3 = 3
log_level = LOG_LEVEL_1
+
def log(level, msg):
if level <= log_level:
print(msg)
+
################################################################################
# Qstr extraction
+
def extract_qstrs(source_files):
def read_qstrs(f):
with open(f) as f:
@@ -191,21 +229,21 @@ def extract_qstrs(source_files):
objs = set()
for line in f:
while line:
- m = re.search(r'MP_OBJ_NEW_QSTR\((MP_QSTR_[A-Za-z0-9_]*)\)', line)
+ m = re.search(r"MP_OBJ_NEW_QSTR\((MP_QSTR_[A-Za-z0-9_]*)\)", line)
if m:
objs.add(m.group(1))
else:
- m = re.search(r'MP_QSTR_[A-Za-z0-9_]*', line)
+ m = re.search(r"MP_QSTR_[A-Za-z0-9_]*", line)
if m:
vals.add(m.group())
if m:
s = m.span()
- line = line[:s[0]] + line[s[1]:]
+ line = line[: s[0]] + line[s[1] :]
else:
- line = ''
+ line = ""
return vals, objs
- static_qstrs = ['MP_QSTR_' + qstrutil.qstr_escape(q) for q in qstrutil.static_qstr_list]
+ static_qstrs = ["MP_QSTR_" + qstrutil.qstr_escape(q) for q in qstrutil.static_qstr_list]
qstr_vals = set()
qstr_objs = set()
@@ -217,12 +255,15 @@ def extract_qstrs(source_files):
return static_qstrs, qstr_vals, qstr_objs
+
################################################################################
# Linker
+
class LinkError(Exception):
pass
+
class Section:
def __init__(self, name, data, alignment, filename=None):
self.filename = filename
@@ -237,6 +278,7 @@ class Section:
assert elfsec.header.sh_addr == 0
return Section(elfsec.name, elfsec.data(), elfsec.data_alignment, filename)
+
class GOTEntry:
def __init__(self, name, sym, link_addr=0):
self.name = name
@@ -245,60 +287,67 @@ class GOTEntry:
self.link_addr = link_addr
def isexternal(self):
- return self.sec_name.startswith('.external')
+ return self.sec_name.startswith(".external")
def istext(self):
- return self.sec_name.startswith('.text')
+ return self.sec_name.startswith(".text")
def isrodata(self):
- return self.sec_name.startswith(('.rodata', '.data.rel.ro'))
+ return self.sec_name.startswith((".rodata", ".data.rel.ro"))
def isbss(self):
- return self.sec_name.startswith('.bss')
+ return self.sec_name.startswith(".bss")
+
class LiteralEntry:
def __init__(self, value, offset):
self.value = value
self.offset = offset
+
class LinkEnv:
def __init__(self, arch):
self.arch = ARCH_DATA[arch]
- self.sections = [] # list of sections in order of output
+ self.sections = [] # list of sections in order of output
self.literal_sections = [] # list of literal sections (xtensa only)
- self.known_syms = {} # dict of symbols that are defined
- self.unresolved_syms = [] # list of unresolved symbols
- self.mpy_relocs = [] # list of relocations needed in the output .mpy file
+ self.known_syms = {} # dict of symbols that are defined
+ self.unresolved_syms = [] # list of unresolved symbols
+ self.mpy_relocs = [] # list of relocations needed in the output .mpy file
def check_arch(self, arch_name):
if arch_name != self.arch.name:
- raise LinkError('incompatible arch')
+ raise LinkError("incompatible arch")
def print_sections(self):
- log(LOG_LEVEL_2, 'sections:')
+ log(LOG_LEVEL_2, "sections:")
for sec in self.sections:
- log(LOG_LEVEL_2, ' {:08x} {} size={}'.format(sec.addr, sec.name, len(sec.data)))
+ log(LOG_LEVEL_2, " {:08x} {} size={}".format(sec.addr, sec.name, len(sec.data)))
def find_addr(self, name):
if name in self.known_syms:
s = self.known_syms[name]
- return s.section.addr + s['st_value']
- raise LinkError('unknown symbol: {}'.format(name))
+ return s.section.addr + s["st_value"]
+ raise LinkError("unknown symbol: {}".format(name))
+
def build_got_generic(env):
env.got_entries = {}
for sec in env.sections:
for r in sec.reloc:
s = r.sym
- if not (s.entry['st_info']['bind'] == 'STB_GLOBAL' and r['r_info_type'] in env.arch.arch_got):
+ if not (
+ s.entry["st_info"]["bind"] == "STB_GLOBAL"
+ and r["r_info_type"] in env.arch.arch_got
+ ):
continue
- s_type = s.entry['st_info']['type']
- assert s_type in ('STT_NOTYPE', 'STT_FUNC', 'STT_OBJECT'), s_type
+ s_type = s.entry["st_info"]["type"]
+ assert s_type in ("STT_NOTYPE", "STT_FUNC", "STT_OBJECT"), s_type
assert s.name
if s.name in env.got_entries:
continue
env.got_entries[s.name] = GOTEntry(s.name, s)
+
def build_got_xtensa(env):
env.got_entries = {}
env.lit_entries = {}
@@ -311,21 +360,21 @@ def build_got_xtensa(env):
# Look through literal relocations to find any global pointers that should be GOT entries
for r in sec.reloc:
s = r.sym
- s_type = s.entry['st_info']['type']
- assert s_type in ('STT_NOTYPE', 'STT_FUNC', 'STT_OBJECT', 'STT_SECTION'), s_type
- assert r['r_info_type'] in env.arch.arch_got
- assert r['r_offset'] % env.arch.word_size == 0
+ s_type = s.entry["st_info"]["type"]
+ assert s_type in ("STT_NOTYPE", "STT_FUNC", "STT_OBJECT", "STT_SECTION"), s_type
+ assert r["r_info_type"] in env.arch.arch_got
+ assert r["r_offset"] % env.arch.word_size == 0
# This entry is a global pointer
- existing = struct.unpack_from('<I', sec.data, r['r_offset'])[0]
- if s_type == 'STT_SECTION':
- assert r['r_addend'] == 0
+ existing = struct.unpack_from("<I", sec.data, r["r_offset"])[0]
+ if s_type == "STT_SECTION":
+ assert r["r_addend"] == 0
name = "{}+0x{:x}".format(s.section.name, existing)
else:
assert existing == 0
name = s.name
- if r['r_addend'] != 0:
- name = "{}+0x{:x}".format(name, r['r_addend'])
- idx = '{}+0x{:x}'.format(sec.filename, r['r_offset'])
+ if r["r_addend"] != 0:
+ name = "{}+0x{:x}".format(name, r["r_addend"])
+ idx = "{}+0x{:x}".format(sec.filename, r["r_offset"])
env.xt_literals[idx] = name
if name in env.got_entries:
# Deduplicate GOT entries
@@ -334,30 +383,35 @@ def build_got_xtensa(env):
# Go through all literal entries finding those that aren't global pointers so must be actual literals
for i in range(0, len(sec.data), env.arch.word_size):
- idx = '{}+0x{:x}'.format(sec.filename, i)
+ idx = "{}+0x{:x}".format(sec.filename, i)
if idx not in env.xt_literals:
# This entry is an actual literal
- value = struct.unpack_from('<I', sec.data, i)[0]
+ value = struct.unpack_from("<I", sec.data, i)[0]
env.xt_literals[idx] = value
if value in env.lit_entries:
# Deduplicate literals
continue
- env.lit_entries[value] = LiteralEntry(value, len(env.lit_entries) * env.arch.word_size)
+ env.lit_entries[value] = LiteralEntry(
+ value, len(env.lit_entries) * env.arch.word_size
+ )
+
def populate_got(env):
# Compute GOT destination addresses
for got_entry in env.got_entries.values():
sym = got_entry.sym
- if hasattr(sym, 'resolved'):
+ if hasattr(sym, "resolved"):
sym = sym.resolved
sec = sym.section
- addr = sym['st_value']
+ addr = sym["st_value"]
got_entry.sec_name = sec.name
got_entry.link_addr += sec.addr + addr
# Get sorted GOT, sorted by external, text, rodata, bss so relocations can be combined
- got_list = sorted(env.got_entries.values(),
- key=lambda g: g.isexternal() + 2 * g.istext() + 3 * g.isrodata() + 4 * g.isbss())
+ got_list = sorted(
+ env.got_entries.values(),
+ key=lambda g: g.isexternal() + 2 * g.istext() + 3 * g.isrodata() + 4 * g.isbss(),
+ )
# Layout and populate the GOT
offset = 0
@@ -365,131 +419,154 @@ def populate_got(env):
got_entry.offset = offset
offset += env.arch.word_size
o = env.got_section.addr + got_entry.offset
- env.full_text[o:o + env.arch.word_size] = got_entry.link_addr.to_bytes(env.arch.word_size, 'little')
+ env.full_text[o : o + env.arch.word_size] = got_entry.link_addr.to_bytes(
+ env.arch.word_size, "little"
+ )
# Create a relocation for each GOT entry
for got_entry in got_list:
- if got_entry.name == 'mp_fun_table':
- dest = 'mp_fun_table'
- elif got_entry.name.startswith('mp_fun_table+0x'):
- dest = int(got_entry.name.split('+')[1], 16) // env.arch.word_size
- elif got_entry.sec_name.startswith('.text'):
- dest = '.text'
- elif got_entry.sec_name.startswith('.rodata'):
- dest = '.rodata'
- elif got_entry.sec_name.startswith('.data.rel.ro'):
- dest = '.data.rel.ro'
- elif got_entry.sec_name.startswith('.bss'):
- dest = '.bss'
+ if got_entry.name == "mp_fun_table":
+ dest = "mp_fun_table"
+ elif got_entry.name.startswith("mp_fun_table+0x"):
+ dest = int(got_entry.name.split("+")[1], 16) // env.arch.word_size
+ elif got_entry.sec_name.startswith(".text"):
+ dest = ".text"
+ elif got_entry.sec_name.startswith(".rodata"):
+ dest = ".rodata"
+ elif got_entry.sec_name.startswith(".data.rel.ro"):
+ dest = ".data.rel.ro"
+ elif got_entry.sec_name.startswith(".bss"):
+ dest = ".bss"
else:
assert 0, (got_entry.name, got_entry.sec_name)
- env.mpy_relocs.append(('.text', env.got_section.addr + got_entry.offset, dest))
+ env.mpy_relocs.append((".text", env.got_section.addr + got_entry.offset, dest))
# Print out the final GOT
- log(LOG_LEVEL_2, 'GOT: {:08x}'.format(env.got_section.addr))
+ log(LOG_LEVEL_2, "GOT: {:08x}".format(env.got_section.addr))
for g in got_list:
- log(LOG_LEVEL_2, ' {:08x} {} -> {}+{:08x}'.format(g.offset, g.name, g.sec_name, g.link_addr))
+ log(
+ LOG_LEVEL_2,
+ " {:08x} {} -> {}+{:08x}".format(g.offset, g.name, g.sec_name, g.link_addr),
+ )
+
def populate_lit(env):
- log(LOG_LEVEL_2, 'LIT: {:08x}'.format(env.lit_section.addr))
+ log(LOG_LEVEL_2, "LIT: {:08x}".format(env.lit_section.addr))
for lit_entry in env.lit_entries.values():
value = lit_entry.value
- log(LOG_LEVEL_2, ' {:08x} = {:08x}'.format(lit_entry.offset, value))
+ log(LOG_LEVEL_2, " {:08x} = {:08x}".format(lit_entry.offset, value))
o = env.lit_section.addr + lit_entry.offset
- env.full_text[o:o + env.arch.word_size] = value.to_bytes(env.arch.word_size, 'little')
+ env.full_text[o : o + env.arch.word_size] = value.to_bytes(env.arch.word_size, "little")
+
def do_relocation_text(env, text_addr, r):
# Extract relevant info about symbol that's being relocated
s = r.sym
- s_bind = s.entry['st_info']['bind']
- s_shndx = s.entry['st_shndx']
- s_type = s.entry['st_info']['type']
- r_offset = r['r_offset'] + text_addr
- r_info_type = r['r_info_type']
+ s_bind = s.entry["st_info"]["bind"]
+ s_shndx = s.entry["st_shndx"]
+ s_type = s.entry["st_info"]["type"]
+ r_offset = r["r_offset"] + text_addr
+ r_info_type = r["r_info_type"]
try:
# only for RELA sections
- r_addend = r['r_addend']
+ r_addend = r["r_addend"]
except KeyError:
r_addend = 0
# Default relocation type and name for logging
- reloc_type = 'le32'
+ reloc_type = "le32"
log_name = None
- if (env.arch.name == 'EM_386' and r_info_type in (R_386_PC32, R_386_PLT32)
- or env.arch.name == 'EM_X86_64' and r_info_type in (R_X86_64_PC32, R_X86_64_PLT32)
- or env.arch.name == 'EM_ARM' and r_info_type in (R_ARM_REL32, R_ARM_THM_CALL, R_ARM_THM_JUMP24)
- or s_bind == 'STB_LOCAL' and env.arch.name == 'EM_XTENSA' and r_info_type == R_XTENSA_32 # not GOT
- ):
+ if (
+ env.arch.name == "EM_386"
+ and r_info_type in (R_386_PC32, R_386_PLT32)
+ or env.arch.name == "EM_X86_64"
+ and r_info_type in (R_X86_64_PC32, R_X86_64_PLT32)
+ or env.arch.name == "EM_ARM"
+ and r_info_type in (R_ARM_REL32, R_ARM_THM_CALL, R_ARM_THM_JUMP24)
+ or s_bind == "STB_LOCAL"
+ and env.arch.name == "EM_XTENSA"
+ and r_info_type == R_XTENSA_32 # not GOT
+ ):
# Standard relocation to fixed location within text/rodata
- if hasattr(s, 'resolved'):
+ if hasattr(s, "resolved"):
s = s.resolved
sec = s.section
- if env.arch.separate_rodata and sec.name.startswith('.rodata'):
- raise LinkError('fixed relocation to rodata with rodata referenced via GOT')
+ if env.arch.separate_rodata and sec.name.startswith(".rodata"):
+ raise LinkError("fixed relocation to rodata with rodata referenced via GOT")
- if sec.name.startswith('.bss'):
- raise LinkError('{}: fixed relocation to bss (bss variables can\'t be static)'.format(s.filename))
+ if sec.name.startswith(".bss"):
+ raise LinkError(
+ "{}: fixed relocation to bss (bss variables can't be static)".format(s.filename)
+ )
- if sec.name.startswith('.external'):
- raise LinkError('{}: fixed relocation to external symbol: {}'.format(s.filename, s.name))
+ if sec.name.startswith(".external"):
+ raise LinkError(
+ "{}: fixed relocation to external symbol: {}".format(s.filename, s.name)
+ )
- addr = sec.addr + s['st_value']
+ addr = sec.addr + s["st_value"]
reloc = addr - r_offset + r_addend
if r_info_type in (R_ARM_THM_CALL, R_ARM_THM_JUMP24):
# Both relocations have the same bit pattern to rewrite:
# R_ARM_THM_CALL: bl
# R_ARM_THM_JUMP24: b.w
- reloc_type = 'thumb_b'
-
- elif (env.arch.name == 'EM_386' and r_info_type == R_386_GOTPC
- or env.arch.name == 'EM_ARM' and r_info_type == R_ARM_BASE_PREL
- ):
+ reloc_type = "thumb_b"
+
+ elif (
+ env.arch.name == "EM_386"
+ and r_info_type == R_386_GOTPC
+ or env.arch.name == "EM_ARM"
+ and r_info_type == R_ARM_BASE_PREL
+ ):
# Relocation to GOT address itself
- assert s.name == '_GLOBAL_OFFSET_TABLE_'
+ assert s.name == "_GLOBAL_OFFSET_TABLE_"
addr = env.got_section.addr
reloc = addr - r_offset + r_addend
- elif (env.arch.name == 'EM_386' and r_info_type in (R_386_GOT32, R_386_GOT32X)
- or env.arch.name == 'EM_ARM' and r_info_type == R_ARM_GOT_BREL
- ):
+ elif (
+ env.arch.name == "EM_386"
+ and r_info_type in (R_386_GOT32, R_386_GOT32X)
+ or env.arch.name == "EM_ARM"
+ and r_info_type == R_ARM_GOT_BREL
+ ):
# Relcation pointing to GOT
reloc = addr = env.got_entries[s.name].offset
- elif env.arch.name == 'EM_X86_64' and r_info_type == R_X86_64_REX_GOTPCRELX:
+ elif env.arch.name == "EM_X86_64" and r_info_type == R_X86_64_REX_GOTPCRELX:
# Relcation pointing to GOT
got_entry = env.got_entries[s.name]
addr = env.got_section.addr + got_entry.offset
reloc = addr - r_offset + r_addend
- elif env.arch.name == 'EM_386' and r_info_type == R_386_GOTOFF:
+ elif env.arch.name == "EM_386" and r_info_type == R_386_GOTOFF:
# Relocation relative to GOT
- addr = s.section.addr + s['st_value']
+ addr = s.section.addr + s["st_value"]
reloc = addr - env.got_section.addr + r_addend
- elif env.arch.name == 'EM_XTENSA' and r_info_type == R_XTENSA_SLOT0_OP:
+ elif env.arch.name == "EM_XTENSA" and r_info_type == R_XTENSA_SLOT0_OP:
# Relocation pointing to GOT, xtensa specific
sec = s.section
- if sec.name.startswith('.text'):
+ if sec.name.startswith(".text"):
# it looks like R_XTENSA_SLOT0_OP into .text is already correctly relocated
return
- assert sec.name.startswith('.literal'), sec.name
- lit_idx = '{}+0x{:x}'.format(sec.filename, r_addend)
+ assert sec.name.startswith(".literal"), sec.name
+ lit_idx = "{}+0x{:x}".format(sec.filename, r_addend)
lit_ptr = env.xt_literals[lit_idx]
if isinstance(lit_ptr, str):
addr = env.got_section.addr + env.got_entries[lit_ptr].offset
- log_name = 'GOT {}'.format(lit_ptr)
+ log_name = "GOT {}".format(lit_ptr)
else:
addr = env.lit_section.addr + env.lit_entries[lit_ptr].offset
- log_name = 'LIT'
+ log_name = "LIT"
reloc = addr - r_offset
- reloc_type = 'xtensa_l32r'
+ reloc_type = "xtensa_l32r"
- elif env.arch.name == 'EM_XTENSA' and r_info_type == R_XTENSA_DIFF32:
- if s.section.name.startswith('.text'):
+ elif env.arch.name == "EM_XTENSA" and r_info_type == R_XTENSA_DIFF32:
+ if s.section.name.startswith(".text"):
# it looks like R_XTENSA_DIFF32 into .text is already correctly relocated
return
assert 0
@@ -499,196 +576,216 @@ def do_relocation_text(env, text_addr, r):
assert 0, r_info_type
# Write relocation
- if reloc_type == 'le32':
- existing, = struct.unpack_from('<I', env.full_text, r_offset)
- struct.pack_into('<I', env.full_text, r_offset, (existing + reloc) & 0xffffffff)
- elif reloc_type == 'thumb_b':
- b_h, b_l = struct.unpack_from('<HH', env.full_text, r_offset)
- existing = (b_h & 0x7ff) << 12 | (b_l & 0x7ff) << 1
- if existing >= 0x400000: # 2's complement
+ if reloc_type == "le32":
+ (existing,) = struct.unpack_from("<I", env.full_text, r_offset)
+ struct.pack_into("<I", env.full_text, r_offset, (existing + reloc) & 0xFFFFFFFF)
+ elif reloc_type == "thumb_b":
+ b_h, b_l = struct.unpack_from("<HH", env.full_text, r_offset)
+ existing = (b_h & 0x7FF) << 12 | (b_l & 0x7FF) << 1
+ if existing >= 0x400000: # 2's complement
existing -= 0x800000
new = existing + reloc
- b_h = (b_h & 0xf800) | (new >> 12) & 0x7ff
- b_l = (b_l & 0xf800) | (new >> 1) & 0x7ff
- struct.pack_into('<HH', env.full_text, r_offset, b_h, b_l)
- elif reloc_type == 'xtensa_l32r':
+ b_h = (b_h & 0xF800) | (new >> 12) & 0x7FF
+ b_l = (b_l & 0xF800) | (new >> 1) & 0x7FF
+ struct.pack_into("<HH", env.full_text, r_offset, b_h, b_l)
+ elif reloc_type == "xtensa_l32r":
l32r = unpack_u24le(env.full_text, r_offset)
- assert l32r & 0xf == 1 # RI16 encoded l32r
+ assert l32r & 0xF == 1 # RI16 encoded l32r
l32r_imm16 = l32r >> 8
- l32r_imm16 = (l32r_imm16 + reloc >> 2) & 0xffff
- l32r = l32r & 0xff | l32r_imm16 << 8
+ l32r_imm16 = (l32r_imm16 + reloc >> 2) & 0xFFFF
+ l32r = l32r & 0xFF | l32r_imm16 << 8
pack_u24le(env.full_text, r_offset, l32r)
else:
assert 0, reloc_type
# Log information about relocation
if log_name is None:
- if s_type == 'STT_SECTION':
+ if s_type == "STT_SECTION":
log_name = s.section.name
else:
log_name = s.name
- log(LOG_LEVEL_3, ' {:08x} {} -> {:08x}'.format(r_offset, log_name, addr))
+ log(LOG_LEVEL_3, " {:08x} {} -> {:08x}".format(r_offset, log_name, addr))
+
def do_relocation_data(env, text_addr, r):
s = r.sym
- s_type = s.entry['st_info']['type']
- r_offset = r['r_offset'] + text_addr
- r_info_type = r['r_info_type']
+ s_type = s.entry["st_info"]["type"]
+ r_offset = r["r_offset"] + text_addr
+ r_info_type = r["r_info_type"]
try:
# only for RELA sections
- r_addend = r['r_addend']
+ r_addend = r["r_addend"]
except KeyError:
r_addend = 0
- if (env.arch.name == 'EM_386' and r_info_type == R_386_32
- or env.arch.name == 'EM_X86_64' and r_info_type == R_X86_64_64
- or env.arch.name == 'EM_ARM' and r_info_type == R_ARM_ABS32
- or env.arch.name == 'EM_XTENSA' and r_info_type == R_XTENSA_32):
+ if (
+ env.arch.name == "EM_386"
+ and r_info_type == R_386_32
+ or env.arch.name == "EM_X86_64"
+ and r_info_type == R_X86_64_64
+ or env.arch.name == "EM_ARM"
+ and r_info_type == R_ARM_ABS32
+ or env.arch.name == "EM_XTENSA"
+ and r_info_type == R_XTENSA_32
+ ):
# Relocation in data.rel.ro to internal/external symbol
if env.arch.word_size == 4:
- struct_type = '<I'
+ struct_type = "<I"
elif env.arch.word_size == 8:
- struct_type = '<Q'
+ struct_type = "<Q"
sec = s.section
assert r_offset % env.arch.word_size == 0
- addr = sec.addr + s['st_value'] + r_addend
- if s_type == 'STT_SECTION':
+ addr = sec.addr + s["st_value"] + r_addend
+ if s_type == "STT_SECTION":
log_name = sec.name
else:
log_name = s.name
- log(LOG_LEVEL_3, ' {:08x} -> {} {:08x}'.format(r_offset, log_name, addr))
+ log(LOG_LEVEL_3, " {:08x} -> {} {:08x}".format(r_offset, log_name, addr))
if env.arch.separate_rodata:
data = env.full_rodata
else:
data = env.full_text
- existing, = struct.unpack_from(struct_type, data, r_offset)
- if sec.name.startswith(('.text', '.rodata', '.data.rel.ro', '.bss')):
+ (existing,) = struct.unpack_from(struct_type, data, r_offset)
+ if sec.name.startswith((".text", ".rodata", ".data.rel.ro", ".bss")):
struct.pack_into(struct_type, data, r_offset, existing + addr)
kind = sec.name
- elif sec.name == '.external.mp_fun_table':
+ elif sec.name == ".external.mp_fun_table":
assert addr == 0
kind = s.mp_fun_table_offset
else:
assert 0, sec.name
if env.arch.separate_rodata:
- base = '.rodata'
+ base = ".rodata"
else:
- base = '.text'
+ base = ".text"
env.mpy_relocs.append((base, r_offset, kind))
else:
# Unknown/unsupported relocation
assert 0, r_info_type
+
def load_object_file(env, felf):
- with open(felf, 'rb') as f:
+ with open(felf, "rb") as f:
elf = elffile.ELFFile(f)
- env.check_arch(elf['e_machine'])
+ env.check_arch(elf["e_machine"])
# Get symbol table
- symtab = list(elf.get_section_by_name('.symtab').iter_symbols())
+ symtab = list(elf.get_section_by_name(".symtab").iter_symbols())
# Load needed sections from ELF file
- sections_shndx = {} # maps elf shndx to Section object
+ sections_shndx = {} # maps elf shndx to Section object
for idx, s in enumerate(elf.iter_sections()):
- if s.header.sh_type in ('SHT_PROGBITS', 'SHT_NOBITS'):
+ if s.header.sh_type in ("SHT_PROGBITS", "SHT_NOBITS"):
if s.data_size == 0:
# Ignore empty sections
pass
- elif s.name.startswith(('.literal', '.text', '.rodata', '.data.rel.ro', '.bss')):
+ elif s.name.startswith((".literal", ".text", ".rodata", ".data.rel.ro", ".bss")):
sec = Section.from_elfsec(s, felf)
sections_shndx[idx] = sec
- if s.name.startswith('.literal'):
+ if s.name.startswith(".literal"):
env.literal_sections.append(sec)
else:
env.sections.append(sec)
- elif s.name.startswith('.data'):
- raise LinkError('{}: {} non-empty'.format(felf, s.name))
+ elif s.name.startswith(".data"):
+ raise LinkError("{}: {} non-empty".format(felf, s.name))
else:
# Ignore section
pass
- elif s.header.sh_type in ('SHT_REL', 'SHT_RELA'):
+ elif s.header.sh_type in ("SHT_REL", "SHT_RELA"):
shndx = s.header.sh_info
if shndx in sections_shndx:
sec = sections_shndx[shndx]
sec.reloc_name = s.name
sec.reloc = list(s.iter_relocations())
for r in sec.reloc:
- r.sym = symtab[r['r_info_sym']]
+ r.sym = symtab[r["r_info_sym"]]
# Link symbols to their sections, and update known and unresolved symbols
for sym in symtab:
sym.filename = felf
- shndx = sym.entry['st_shndx']
+ shndx = sym.entry["st_shndx"]
if shndx in sections_shndx:
# Symbol with associated section
sym.section = sections_shndx[shndx]
- if sym['st_info']['bind'] == 'STB_GLOBAL':
+ if sym["st_info"]["bind"] == "STB_GLOBAL":
# Defined global symbol
- if sym.name in env.known_syms and not sym.name.startswith('__x86.get_pc_thunk.'):
- raise LinkError('duplicate symbol: {}'.format(sym.name))
+ if sym.name in env.known_syms and not sym.name.startswith(
+ "__x86.get_pc_thunk."
+ ):
+ raise LinkError("duplicate symbol: {}".format(sym.name))
env.known_syms[sym.name] = sym
- elif sym.entry['st_shndx'] == 'SHN_UNDEF' and sym['st_info']['bind'] == 'STB_GLOBAL':
+ elif sym.entry["st_shndx"] == "SHN_UNDEF" and sym["st_info"]["bind"] == "STB_GLOBAL":
# Undefined global symbol, needs resolving
env.unresolved_syms.append(sym)
+
def link_objects(env, native_qstr_vals_len, native_qstr_objs_len):
# Build GOT information
- if env.arch.name == 'EM_XTENSA':
+ if env.arch.name == "EM_XTENSA":
build_got_xtensa(env)
else:
build_got_generic(env)
# Creat GOT section
got_size = len(env.got_entries) * env.arch.word_size
- env.got_section = Section('GOT', bytearray(got_size), env.arch.word_size)
- if env.arch.name == 'EM_XTENSA':
+ env.got_section = Section("GOT", bytearray(got_size), env.arch.word_size)
+ if env.arch.name == "EM_XTENSA":
env.sections.insert(0, env.got_section)
else:
env.sections.append(env.got_section)
# Create optional literal section
- if env.arch.name == 'EM_XTENSA':
+ if env.arch.name == "EM_XTENSA":
lit_size = len(env.lit_entries) * env.arch.word_size
- env.lit_section = Section('LIT', bytearray(lit_size), env.arch.word_size)
+ env.lit_section = Section("LIT", bytearray(lit_size), env.arch.word_size)
env.sections.insert(1, env.lit_section)
# Create section to contain mp_native_qstr_val_table
- env.qstr_val_section = Section('.text.QSTR_VAL', bytearray(native_qstr_vals_len * env.arch.qstr_entry_size), env.arch.qstr_entry_size)
+ env.qstr_val_section = Section(
+ ".text.QSTR_VAL",
+ bytearray(native_qstr_vals_len * env.arch.qstr_entry_size),
+ env.arch.qstr_entry_size,
+ )
env.sections.append(env.qstr_val_section)
# Create section to contain mp_native_qstr_obj_table
- env.qstr_obj_section = Section('.text.QSTR_OBJ', bytearray(native_qstr_objs_len * env.arch.word_size), env.arch.word_size)
+ env.qstr_obj_section = Section(
+ ".text.QSTR_OBJ", bytearray(native_qstr_objs_len * env.arch.word_size), env.arch.word_size
+ )
env.sections.append(env.qstr_obj_section)
# Resolve unknown symbols
- mp_fun_table_sec = Section('.external.mp_fun_table', b'', 0)
- fun_table = {key: 67 + idx
- for idx, key in enumerate([
- 'mp_type_type',
- 'mp_type_str',
- 'mp_type_list',
- 'mp_type_dict',
- 'mp_type_fun_builtin_0',
- 'mp_type_fun_builtin_1',
- 'mp_type_fun_builtin_2',
- 'mp_type_fun_builtin_3',
- 'mp_type_fun_builtin_var',
- 'mp_stream_read_obj',
- 'mp_stream_readinto_obj',
- 'mp_stream_unbuffered_readline_obj',
- 'mp_stream_write_obj',
- ])
+ mp_fun_table_sec = Section(".external.mp_fun_table", b"", 0)
+ fun_table = {
+ key: 67 + idx
+ for idx, key in enumerate(
+ [
+ "mp_type_type",
+ "mp_type_str",
+ "mp_type_list",
+ "mp_type_dict",
+ "mp_type_fun_builtin_0",
+ "mp_type_fun_builtin_1",
+ "mp_type_fun_builtin_2",
+ "mp_type_fun_builtin_3",
+ "mp_type_fun_builtin_var",
+ "mp_stream_read_obj",
+ "mp_stream_readinto_obj",
+ "mp_stream_unbuffered_readline_obj",
+ "mp_stream_write_obj",
+ ]
+ )
}
for sym in env.unresolved_syms:
- assert sym['st_value'] == 0
- if sym.name == '_GLOBAL_OFFSET_TABLE_':
+ assert sym["st_value"] == 0
+ if sym.name == "_GLOBAL_OFFSET_TABLE_":
pass
- elif sym.name == 'mp_fun_table':
- sym.section = Section('.external', b'', 0)
- elif sym.name == 'mp_native_qstr_val_table':
+ elif sym.name == "mp_fun_table":
+ sym.section = Section(".external", b"", 0)
+ elif sym.name == "mp_native_qstr_val_table":
sym.section = env.qstr_val_section
- elif sym.name == 'mp_native_qstr_obj_table':
+ elif sym.name == "mp_native_qstr_obj_table":
sym.section = env.qstr_obj_section
elif sym.name in env.known_syms:
sym.resolved = env.known_syms[sym.name]
@@ -697,48 +794,53 @@ def link_objects(env, native_qstr_vals_len, native_qstr_objs_len):
sym.section = mp_fun_table_sec
sym.mp_fun_table_offset = fun_table[sym.name]
else:
- raise LinkError('{}: undefined symbol: {}'.format(sym.filename, sym.name))
+ raise LinkError("{}: undefined symbol: {}".format(sym.filename, sym.name))
# Align sections, assign their addresses, and create full_text
- env.full_text = bytearray(env.arch.asm_jump(8)) # dummy, to be filled in later
+ env.full_text = bytearray(env.arch.asm_jump(8)) # dummy, to be filled in later
env.full_rodata = bytearray(0)
env.full_bss = bytearray(0)
for sec in env.sections:
- if env.arch.separate_rodata and sec.name.startswith(('.rodata', '.data.rel.ro')):
+ if env.arch.separate_rodata and sec.name.startswith((".rodata", ".data.rel.ro")):
data = env.full_rodata
- elif sec.name.startswith('.bss'):
+ elif sec.name.startswith(".bss"):
data = env.full_bss
else:
data = env.full_text
sec.addr = align_to(len(data), sec.alignment)
- data.extend(b'\x00' * (sec.addr - len(data)))
+ data.extend(b"\x00" * (sec.addr - len(data)))
data.extend(sec.data)
env.print_sections()
populate_got(env)
- if env.arch.name == 'EM_XTENSA':
+ if env.arch.name == "EM_XTENSA":
populate_lit(env)
# Fill in relocations
for sec in env.sections:
if not sec.reloc:
continue
- log(LOG_LEVEL_3, '{}: {} relocations via {}:'.format(sec.filename, sec.name, sec.reloc_name))
+ log(
+ LOG_LEVEL_3,
+ "{}: {} relocations via {}:".format(sec.filename, sec.name, sec.reloc_name),
+ )
for r in sec.reloc:
- if sec.name.startswith(('.text', '.rodata')):
+ if sec.name.startswith((".text", ".rodata")):
do_relocation_text(env, sec.addr, r)
- elif sec.name.startswith('.data.rel.ro'):
+ elif sec.name.startswith(".data.rel.ro"):
do_relocation_data(env, sec.addr, r)
else:
assert 0, sec.name
+
################################################################################
# .mpy output
+
class MPYOutput:
def open(self, fname):
- self.f = open(fname, 'wb')
+ self.f = open(fname, "wb")
self.prev_base = -1
self.prev_offset = -1
@@ -750,10 +852,10 @@ class MPYOutput:
def write_uint(self, val):
b = bytearray()
- b.insert(0, val & 0x7f)
+ b.insert(0, val & 0x7F)
val >>= 7
while val:
- b.insert(0, 0x80 | (val & 0x7f))
+ b.insert(0, 0x80 | (val & 0x7F))
val >>= 7
self.write_bytes(b)
@@ -761,7 +863,7 @@ class MPYOutput:
if s in qstrutil.static_qstr_list:
self.write_bytes(bytes([0, qstrutil.static_qstr_list.index(s) + 1]))
else:
- s = bytes(s, 'ascii')
+ s = bytes(s, "ascii")
self.write_uint(len(s) << 1)
self.write_bytes(s)
@@ -774,42 +876,41 @@ class MPYOutput:
assert 6 <= dest <= 127
assert n == 1
dest = dest << 1 | need_offset
- assert 0 <= dest <= 0xfe, dest
+ assert 0 <= dest <= 0xFE, dest
self.write_bytes(bytes([dest]))
if need_offset:
- if base == '.text':
+ if base == ".text":
base = 0
- elif base == '.rodata':
+ elif base == ".rodata":
base = 1
self.write_uint(offset << 1 | base)
if n > 1:
self.write_uint(n)
+
def build_mpy(env, entry_offset, fmpy, native_qstr_vals, native_qstr_objs):
# Write jump instruction to start of text
jump = env.arch.asm_jump(entry_offset)
- env.full_text[:len(jump)] = jump
+ env.full_text[: len(jump)] = jump
- log(LOG_LEVEL_1, 'arch: {}'.format(env.arch.name))
- log(LOG_LEVEL_1, 'text size: {}'.format(len(env.full_text)))
+ log(LOG_LEVEL_1, "arch: {}".format(env.arch.name))
+ log(LOG_LEVEL_1, "text size: {}".format(len(env.full_text)))
if len(env.full_rodata):
- log(LOG_LEVEL_1, 'rodata size: {}'.format(len(env.full_rodata)))
- log(LOG_LEVEL_1, 'bss size: {}'.format(len(env.full_bss)))
- log(LOG_LEVEL_1, 'GOT entries: {}'.format(len(env.got_entries)))
+ log(LOG_LEVEL_1, "rodata size: {}".format(len(env.full_rodata)))
+ log(LOG_LEVEL_1, "bss size: {}".format(len(env.full_bss)))
+ log(LOG_LEVEL_1, "GOT entries: {}".format(len(env.got_entries)))
- #xxd(env.full_text)
+ # xxd(env.full_text)
out = MPYOutput()
out.open(fmpy)
# MPY: header
- out.write_bytes(bytearray([
- ord('M'),
- MPY_VERSION,
- env.arch.mpy_feature,
- MP_SMALL_INT_BITS,
- QSTR_WINDOW_SIZE,
- ]))
+ out.write_bytes(
+ bytearray(
+ [ord("M"), MPY_VERSION, env.arch.mpy_feature, MP_SMALL_INT_BITS, QSTR_WINDOW_SIZE,]
+ )
+ )
# MPY: kind/len
out.write_uint(len(env.full_text) << 2 | (MP_CODE_NATIVE_VIPER - MP_CODE_BYTECODE))
@@ -854,16 +955,16 @@ def build_mpy(env, entry_offset, fmpy, native_qstr_vals, native_qstr_objs):
# MPY: relocation information
prev_kind = None
for base, addr, kind in env.mpy_relocs:
- if isinstance(kind, str) and kind.startswith('.text'):
+ if isinstance(kind, str) and kind.startswith(".text"):
kind = 0
- elif kind in ('.rodata', '.data.rel.ro'):
+ elif kind in (".rodata", ".data.rel.ro"):
if env.arch.separate_rodata:
kind = rodata_const_table_idx
else:
kind = 0
- elif isinstance(kind, str) and kind.startswith('.bss'):
+ elif isinstance(kind, str) and kind.startswith(".bss"):
kind = bss_const_table_idx
- elif kind == 'mp_fun_table':
+ elif kind == "mp_fun_table":
kind = 6
else:
kind = 7 + kind
@@ -883,73 +984,88 @@ def build_mpy(env, entry_offset, fmpy, native_qstr_vals, native_qstr_objs):
out.write_reloc(prev_base, prev_offset - prev_n + 1, prev_kind, prev_n)
# MPY: sentinel for end of relocations
- out.write_bytes(b'\xff')
+ out.write_bytes(b"\xff")
out.close()
+
################################################################################
# main
+
def do_preprocess(args):
if args.output is None:
- assert args.files[0].endswith('.c')
- args.output = args.files[0][:-1] + 'config.h'
+ assert args.files[0].endswith(".c")
+ args.output = args.files[0][:-1] + "config.h"
static_qstrs, qstr_vals, qstr_objs = extract_qstrs(args.files)
- with open(args.output, 'w') as f:
- print('#include <stdint.h>\n'
- 'typedef uintptr_t mp_uint_t;\n'
- 'typedef intptr_t mp_int_t;\n'
- 'typedef uintptr_t mp_off_t;', file=f)
+ with open(args.output, "w") as f:
+ print(
+ "#include <stdint.h>\n"
+ "typedef uintptr_t mp_uint_t;\n"
+ "typedef intptr_t mp_int_t;\n"
+ "typedef uintptr_t mp_off_t;",
+ file=f,
+ )
for i, q in enumerate(static_qstrs):
- print('#define %s (%u)' % (q, i + 1), file=f)
+ print("#define %s (%u)" % (q, i + 1), file=f)
for i, q in enumerate(sorted(qstr_vals)):
- print('#define %s (mp_native_qstr_val_table[%d])' % (q, i), file=f)
+ print("#define %s (mp_native_qstr_val_table[%d])" % (q, i), file=f)
for i, q in enumerate(sorted(qstr_objs)):
- print('#define MP_OBJ_NEW_QSTR_%s ((mp_obj_t)mp_native_qstr_obj_table[%d])' % (q, i), file=f)
- if args.arch == 'xtensawin':
- qstr_type = 'uint32_t' # esp32 can only read 32-bit values from IRAM
+ print(
+ "#define MP_OBJ_NEW_QSTR_%s ((mp_obj_t)mp_native_qstr_obj_table[%d])" % (q, i),
+ file=f,
+ )
+ if args.arch == "xtensawin":
+ qstr_type = "uint32_t" # esp32 can only read 32-bit values from IRAM
else:
- qstr_type = 'uint16_t'
- print('extern const {} mp_native_qstr_val_table[];'.format(qstr_type), file=f)
- print('extern const mp_uint_t mp_native_qstr_obj_table[];', file=f)
+ qstr_type = "uint16_t"
+ print("extern const {} mp_native_qstr_val_table[];".format(qstr_type), file=f)
+ print("extern const mp_uint_t mp_native_qstr_obj_table[];", file=f)
+
def do_link(args):
if args.output is None:
- assert args.files[0].endswith('.o')
- args.output = args.files[0][:-1] + 'mpy'
+ assert args.files[0].endswith(".o")
+ args.output = args.files[0][:-1] + "mpy"
native_qstr_vals = []
native_qstr_objs = []
if args.qstrs is not None:
with open(args.qstrs) as f:
for l in f:
- m = re.match(r'#define MP_QSTR_([A-Za-z0-9_]*) \(mp_native_', l)
+ m = re.match(r"#define MP_QSTR_([A-Za-z0-9_]*) \(mp_native_", l)
if m:
native_qstr_vals.append(m.group(1))
else:
- m = re.match(r'#define MP_OBJ_NEW_QSTR_MP_QSTR_([A-Za-z0-9_]*)', l)
+ m = re.match(r"#define MP_OBJ_NEW_QSTR_MP_QSTR_([A-Za-z0-9_]*)", l)
if m:
native_qstr_objs.append(m.group(1))
- log(LOG_LEVEL_2, 'qstr vals: ' + ', '.join(native_qstr_vals))
- log(LOG_LEVEL_2, 'qstr objs: ' + ', '.join(native_qstr_objs))
+ log(LOG_LEVEL_2, "qstr vals: " + ", ".join(native_qstr_vals))
+ log(LOG_LEVEL_2, "qstr objs: " + ", ".join(native_qstr_objs))
env = LinkEnv(args.arch)
try:
for file in args.files:
load_object_file(env, file)
link_objects(env, len(native_qstr_vals), len(native_qstr_objs))
- build_mpy(env, env.find_addr('mpy_init'), args.output, native_qstr_vals, native_qstr_objs)
+ build_mpy(env, env.find_addr("mpy_init"), args.output, native_qstr_vals, native_qstr_objs)
except LinkError as er:
- print('LinkError:', er.args[0])
+ print("LinkError:", er.args[0])
sys.exit(1)
+
def main():
import argparse
- cmd_parser = argparse.ArgumentParser(description='Run scripts on the pyboard.')
- cmd_parser.add_argument('--verbose', '-v', action='count', default=1, help='increase verbosity')
- cmd_parser.add_argument('--arch', default='x64', help='architecture')
- cmd_parser.add_argument('--preprocess', action='store_true', help='preprocess source files')
- cmd_parser.add_argument('--qstrs', default=None, help='file defining additional qstrs')
- cmd_parser.add_argument('--output', '-o', default=None, help='output .mpy file (default to input with .o->.mpy)')
- cmd_parser.add_argument('files', nargs='+', help='input files')
+
+ cmd_parser = argparse.ArgumentParser(description="Run scripts on the pyboard.")
+ cmd_parser.add_argument(
+ "--verbose", "-v", action="count", default=1, help="increase verbosity"
+ )
+ cmd_parser.add_argument("--arch", default="x64", help="architecture")
+ cmd_parser.add_argument("--preprocess", action="store_true", help="preprocess source files")
+ cmd_parser.add_argument("--qstrs", default=None, help="file defining additional qstrs")
+ cmd_parser.add_argument(
+ "--output", "-o", default=None, help="output .mpy file (default to input with .o->.mpy)"
+ )
+ cmd_parser.add_argument("files", nargs="+", help="input files")
args = cmd_parser.parse_args()
global log_level
@@ -960,5 +1076,6 @@ def main():
else:
do_link(args)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
diff --git a/tools/pyboard.py b/tools/pyboard.py
index a0b4ac320..de8b48836 100755
--- a/tools/pyboard.py
+++ b/tools/pyboard.py
@@ -77,35 +77,42 @@ except AttributeError:
# Python2 doesn't have buffer attr
stdout = sys.stdout
+
def stdout_write_bytes(b):
b = b.replace(b"\x04", b"")
stdout.write(b)
stdout.flush()
+
class PyboardError(Exception):
pass
+
class TelnetToSerial:
def __init__(self, ip, user, password, read_timeout=None):
self.tn = None
import telnetlib
+
self.tn = telnetlib.Telnet(ip, timeout=15)
self.read_timeout = read_timeout
- if b'Login as:' in self.tn.read_until(b'Login as:', timeout=read_timeout):
- self.tn.write(bytes(user, 'ascii') + b"\r\n")
+ if b"Login as:" in self.tn.read_until(b"Login as:", timeout=read_timeout):
+ self.tn.write(bytes(user, "ascii") + b"\r\n")
- if b'Password:' in self.tn.read_until(b'Password:', timeout=read_timeout):
+ if b"Password:" in self.tn.read_until(b"Password:", timeout=read_timeout):
# needed because of internal implementation details of the telnet server
time.sleep(0.2)
- self.tn.write(bytes(password, 'ascii') + b"\r\n")
+ self.tn.write(bytes(password, "ascii") + b"\r\n")
- if b'for more information.' in self.tn.read_until(b'Type "help()" for more information.', timeout=read_timeout):
+ if b"for more information." in self.tn.read_until(
+ b'Type "help()" for more information.', timeout=read_timeout
+ ):
# login successful
from collections import deque
+
self.fifo = deque()
return
- raise PyboardError('Failed to establish a telnet connection with the board')
+ raise PyboardError("Failed to establish a telnet connection with the board")
def __del__(self):
self.close()
@@ -127,7 +134,7 @@ class TelnetToSerial:
break
timeout_count += 1
- data = b''
+ data = b""
while len(data) < size and len(self.fifo) > 0:
data += bytes([self.fifo.popleft()])
return data
@@ -151,24 +158,33 @@ class ProcessToSerial:
def __init__(self, cmd):
import subprocess
- self.subp = subprocess.Popen(cmd, bufsize=0, shell=True, preexec_fn=os.setsid,
- stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+
+ self.subp = subprocess.Popen(
+ cmd,
+ bufsize=0,
+ shell=True,
+ preexec_fn=os.setsid,
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ )
# Initially was implemented with selectors, but that adds Python3
# dependency. However, there can be race conditions communicating
# with a particular child process (like QEMU), and selectors may
# still work better in that case, so left inplace for now.
#
- #import selectors
- #self.sel = selectors.DefaultSelector()
- #self.sel.register(self.subp.stdout, selectors.EVENT_READ)
+ # import selectors
+ # self.sel = selectors.DefaultSelector()
+ # self.sel.register(self.subp.stdout, selectors.EVENT_READ)
import select
+
self.poll = select.poll()
self.poll.register(self.subp.stdout.fileno())
def close(self):
import signal
+
os.killpg(os.getpgid(self.subp.pid), signal.SIGTERM)
def read(self, size=1):
@@ -182,7 +198,7 @@ class ProcessToSerial:
return len(data)
def inWaiting(self):
- #res = self.sel.select(0)
+ # res = self.sel.select(0)
res = self.poll.poll(0)
if res:
return 1
@@ -198,8 +214,16 @@ class ProcessPtyToTerminal:
import subprocess
import re
import serial
- self.subp = subprocess.Popen(cmd.split(), bufsize=0, shell=False, preexec_fn=os.setsid,
- stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+ self.subp = subprocess.Popen(
+ cmd.split(),
+ bufsize=0,
+ shell=False,
+ preexec_fn=os.setsid,
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ )
pty_line = self.subp.stderr.readline().decode("utf-8")
m = re.search(r"/dev/pts/[0-9]+", pty_line)
if not m:
@@ -213,6 +237,7 @@ class ProcessPtyToTerminal:
def close(self):
import signal
+
os.killpg(os.getpgid(self.subp.pid), signal.SIGTERM)
def read(self, size=1):
@@ -226,36 +251,37 @@ class ProcessPtyToTerminal:
class Pyboard:
- def __init__(self, device, baudrate=115200, user='micro', password='python', wait=0):
+ def __init__(self, device, baudrate=115200, user="micro", password="python", wait=0):
if device.startswith("exec:"):
- self.serial = ProcessToSerial(device[len("exec:"):])
+ self.serial = ProcessToSerial(device[len("exec:") :])
elif device.startswith("execpty:"):
- self.serial = ProcessPtyToTerminal(device[len("qemupty:"):])
- elif device and device[0].isdigit() and device[-1].isdigit() and device.count('.') == 3:
+ self.serial = ProcessPtyToTerminal(device[len("qemupty:") :])
+ elif device and device[0].isdigit() and device[-1].isdigit() and device.count(".") == 3:
# device looks like an IP address
self.serial = TelnetToSerial(device, user, password, read_timeout=10)
else:
import serial
+
delayed = False
for attempt in range(wait + 1):
try:
self.serial = serial.Serial(device, baudrate=baudrate, interCharTimeout=1)
break
- except (OSError, IOError): # Py2 and Py3 have different errors
+ except (OSError, IOError): # Py2 and Py3 have different errors
if wait == 0:
continue
if attempt == 0:
- sys.stdout.write('Waiting {} seconds for pyboard '.format(wait))
+ sys.stdout.write("Waiting {} seconds for pyboard ".format(wait))
delayed = True
time.sleep(1)
- sys.stdout.write('.')
+ sys.stdout.write(".")
sys.stdout.flush()
else:
if delayed:
- print('')
- raise PyboardError('failed to access ' + device)
+ print("")
+ raise PyboardError("failed to access " + device)
if delayed:
- print('')
+ print("")
def close(self):
self.serial.close()
@@ -287,7 +313,7 @@ class Pyboard:
return data
def enter_raw_repl(self):
- self.serial.write(b'\r\x03\x03') # ctrl-C twice: interrupt any running program
+ self.serial.write(b"\r\x03\x03") # ctrl-C twice: interrupt any running program
# flush input (without relying on serial.flushInput())
n = self.serial.inWaiting()
@@ -295,38 +321,38 @@ class Pyboard:
self.serial.read(n)
n = self.serial.inWaiting()
- self.serial.write(b'\r\x01') # ctrl-A: enter raw REPL
- data = self.read_until(1, b'raw REPL; CTRL-B to exit\r\n>')
- if not data.endswith(b'raw REPL; CTRL-B to exit\r\n>'):
+ self.serial.write(b"\r\x01") # ctrl-A: enter raw REPL
+ data = self.read_until(1, b"raw REPL; CTRL-B to exit\r\n>")
+ if not data.endswith(b"raw REPL; CTRL-B to exit\r\n>"):
print(data)
- raise PyboardError('could not enter raw repl')
+ raise PyboardError("could not enter raw repl")
- self.serial.write(b'\x04') # ctrl-D: soft reset
- data = self.read_until(1, b'soft reboot\r\n')
- if not data.endswith(b'soft reboot\r\n'):
+ self.serial.write(b"\x04") # ctrl-D: soft reset
+ data = self.read_until(1, b"soft reboot\r\n")
+ if not data.endswith(b"soft reboot\r\n"):
print(data)
- raise PyboardError('could not enter raw repl')
+ raise PyboardError("could not enter raw repl")
# By splitting this into 2 reads, it allows boot.py to print stuff,
# which will show up after the soft reboot and before the raw REPL.
- data = self.read_until(1, b'raw REPL; CTRL-B to exit\r\n')
- if not data.endswith(b'raw REPL; CTRL-B to exit\r\n'):
+ data = self.read_until(1, b"raw REPL; CTRL-B to exit\r\n")
+ if not data.endswith(b"raw REPL; CTRL-B to exit\r\n"):
print(data)
- raise PyboardError('could not enter raw repl')
+ raise PyboardError("could not enter raw repl")
def exit_raw_repl(self):
- self.serial.write(b'\r\x02') # ctrl-B: enter friendly REPL
+ self.serial.write(b"\r\x02") # ctrl-B: enter friendly REPL
def follow(self, timeout, data_consumer=None):
# wait for normal output
- data = self.read_until(1, b'\x04', timeout=timeout, data_consumer=data_consumer)
- if not data.endswith(b'\x04'):
- raise PyboardError('timeout waiting for first EOF reception')
+ data = self.read_until(1, b"\x04", timeout=timeout, data_consumer=data_consumer)
+ if not data.endswith(b"\x04"):
+ raise PyboardError("timeout waiting for first EOF reception")
data = data[:-1]
# wait for error output
- data_err = self.read_until(1, b'\x04', timeout=timeout)
- if not data_err.endswith(b'\x04'):
- raise PyboardError('timeout waiting for second EOF reception')
+ data_err = self.read_until(1, b"\x04", timeout=timeout)
+ if not data_err.endswith(b"\x04"):
+ raise PyboardError("timeout waiting for second EOF reception")
data_err = data_err[:-1]
# return normal and error output
@@ -336,67 +362,71 @@ class Pyboard:
if isinstance(command, bytes):
command_bytes = command
else:
- command_bytes = bytes(command, encoding='utf8')
+ command_bytes = bytes(command, encoding="utf8")
# check we have a prompt
- data = self.read_until(1, b'>')
- if not data.endswith(b'>'):
- raise PyboardError('could not enter raw repl')
+ data = self.read_until(1, b">")
+ if not data.endswith(b">"):
+ raise PyboardError("could not enter raw repl")
# write command
for i in range(0, len(command_bytes), 256):
- self.serial.write(command_bytes[i:min(i + 256, len(command_bytes))])
+ self.serial.write(command_bytes[i : min(i + 256, len(command_bytes))])
time.sleep(0.01)
- self.serial.write(b'\x04')
+ self.serial.write(b"\x04")
# check if we could exec command
data = self.serial.read(2)
- if data != b'OK':
- raise PyboardError('could not exec command (response: %r)' % data)
+ if data != b"OK":
+ raise PyboardError("could not exec command (response: %r)" % data)
def exec_raw(self, command, timeout=10, data_consumer=None):
- self.exec_raw_no_follow(command);
+ self.exec_raw_no_follow(command)
return self.follow(timeout, data_consumer)
def eval(self, expression):
- ret = self.exec_('print({})'.format(expression))
+ ret = self.exec_("print({})".format(expression))
ret = ret.strip()
return ret
def exec_(self, command, data_consumer=None):
ret, ret_err = self.exec_raw(command, data_consumer=data_consumer)
if ret_err:
- raise PyboardError('exception', ret, ret_err)
+ raise PyboardError("exception", ret, ret_err)
return ret
def execfile(self, filename):
- with open(filename, 'rb') as f:
+ with open(filename, "rb") as f:
pyfile = f.read()
return self.exec_(pyfile)
def get_time(self):
- t = str(self.eval('pyb.RTC().datetime()'), encoding='utf8')[1:-1].split(', ')
+ t = str(self.eval("pyb.RTC().datetime()"), encoding="utf8")[1:-1].split(", ")
return int(t[4]) * 3600 + int(t[5]) * 60 + int(t[6])
def fs_ls(self, src):
- cmd = "import uos\nfor f in uos.ilistdir(%s):\n" \
- " print('{:12} {}{}'.format(f[3]if len(f)>3 else 0,f[0],'/'if f[1]&0x4000 else ''))" % \
- (("'%s'" % src) if src else '')
+ cmd = (
+ "import uos\nfor f in uos.ilistdir(%s):\n"
+ " print('{:12} {}{}'.format(f[3]if len(f)>3 else 0,f[0],'/'if f[1]&0x4000 else ''))"
+ % (("'%s'" % src) if src else "")
+ )
self.exec_(cmd, data_consumer=stdout_write_bytes)
def fs_cat(self, src, chunk_size=256):
- cmd = "with open('%s') as f:\n while 1:\n" \
+ cmd = (
+ "with open('%s') as f:\n while 1:\n"
" b=f.read(%u)\n if not b:break\n print(b,end='')" % (src, chunk_size)
+ )
self.exec_(cmd, data_consumer=stdout_write_bytes)
def fs_get(self, src, dest, chunk_size=256):
self.exec_("f=open('%s','rb')\nr=f.read" % src)
- with open(dest, 'wb') as f:
+ with open(dest, "wb") as f:
while True:
data = bytearray()
- self.exec_("print(r(%u))" % chunk_size, data_consumer=lambda d:data.extend(d))
- assert data.endswith(b'\r\n\x04')
- data = eval(str(data[:-3], 'ascii'))
+ self.exec_("print(r(%u))" % chunk_size, data_consumer=lambda d: data.extend(d))
+ assert data.endswith(b"\r\n\x04")
+ data = eval(str(data[:-3], "ascii"))
if not data:
break
f.write(data)
@@ -404,15 +434,15 @@ class Pyboard:
def fs_put(self, src, dest, chunk_size=256):
self.exec_("f=open('%s','wb')\nw=f.write" % dest)
- with open(src, 'rb') as f:
+ with open(src, "rb") as f:
while True:
data = f.read(chunk_size)
if not data:
break
if sys.version_info < (3,):
- self.exec_('w(b' + repr(data) + ')')
+ self.exec_("w(b" + repr(data) + ")")
else:
- self.exec_('w(' + repr(data) + ')')
+ self.exec_("w(" + repr(data) + ")")
self.exec_("f.close()")
def fs_mkdir(self, dir):
@@ -424,11 +454,13 @@ class Pyboard:
def fs_rm(self, src):
self.exec_("import uos\nuos.remove('%s')" % src)
+
# in Python2 exec is a keyword so one must use "exec_"
# but for Python3 we want to provide the nicer version "exec"
setattr(Pyboard, "exec", Pyboard.exec_)
-def execfile(filename, device='/dev/ttyACM0', baudrate=115200, user='micro', password='python'):
+
+def execfile(filename, device="/dev/ttyACM0", baudrate=115200, user="micro", password="python"):
pyb = Pyboard(device, baudrate, user, password)
pyb.enter_raw_repl()
output = pyb.execfile(filename)
@@ -436,54 +468,62 @@ def execfile(filename, device='/dev/ttyACM0', baudrate=115200, user='micro', pas
pyb.exit_raw_repl()
pyb.close()
+
def filesystem_command(pyb, args):
def fname_remote(src):
- if src.startswith(':'):
+ if src.startswith(":"):
src = src[1:]
return src
+
def fname_cp_dest(src, dest):
- src = src.rsplit('/', 1)[-1]
- if dest is None or dest == '':
+ src = src.rsplit("/", 1)[-1]
+ if dest is None or dest == "":
dest = src
- elif dest == '.':
- dest = './' + src
- elif dest.endswith('/'):
+ elif dest == ".":
+ dest = "./" + src
+ elif dest.endswith("/"):
dest += src
return dest
cmd = args[0]
args = args[1:]
try:
- if cmd == 'cp':
+ if cmd == "cp":
srcs = args[:-1]
dest = args[-1]
- if srcs[0].startswith('./') or dest.startswith(':'):
+ if srcs[0].startswith("./") or dest.startswith(":"):
op = pyb.fs_put
- fmt = 'cp %s :%s'
+ fmt = "cp %s :%s"
dest = fname_remote(dest)
else:
op = pyb.fs_get
- fmt = 'cp :%s %s'
+ fmt = "cp :%s %s"
for src in srcs:
src = fname_remote(src)
dest2 = fname_cp_dest(src, dest)
print(fmt % (src, dest2))
op(src, dest2)
else:
- op = {'ls': pyb.fs_ls, 'cat': pyb.fs_cat, 'mkdir': pyb.fs_mkdir,
- 'rmdir': pyb.fs_rmdir, 'rm': pyb.fs_rm}[cmd]
- if cmd == 'ls' and not args:
- args = ['']
+ op = {
+ "ls": pyb.fs_ls,
+ "cat": pyb.fs_cat,
+ "mkdir": pyb.fs_mkdir,
+ "rmdir": pyb.fs_rmdir,
+ "rm": pyb.fs_rm,
+ }[cmd]
+ if cmd == "ls" and not args:
+ args = [""]
for src in args:
src = fname_remote(src)
- print('%s :%s' % (cmd, src))
+ print("%s :%s" % (cmd, src))
op(src)
except PyboardError as er:
- print(str(er.args[2], 'ascii'))
+ print(str(er.args[2], "ascii"))
pyb.exit_raw_repl()
pyb.close()
sys.exit(1)
+
_injected_import_hook_code = """\
import uos, uio
class _FS:
@@ -511,20 +551,44 @@ uos.umount('/_')
del _injected_buf, _FS
"""
+
def main():
import argparse
- cmd_parser = argparse.ArgumentParser(description='Run scripts on the pyboard.')
- cmd_parser.add_argument('--device', default='/dev/ttyACM0', help='the serial device or the IP address of the pyboard')
- cmd_parser.add_argument('-b', '--baudrate', default=115200, help='the baud rate of the serial device')
- cmd_parser.add_argument('-u', '--user', default='micro', help='the telnet login username')
- cmd_parser.add_argument('-p', '--password', default='python', help='the telnet login password')
- cmd_parser.add_argument('-c', '--command', help='program passed in as string')
- cmd_parser.add_argument('-w', '--wait', default=0, type=int, help='seconds to wait for USB connected board to become available')
+
+ cmd_parser = argparse.ArgumentParser(description="Run scripts on the pyboard.")
+ cmd_parser.add_argument(
+ "--device",
+ default="/dev/ttyACM0",
+ help="the serial device or the IP address of the pyboard",
+ )
+ cmd_parser.add_argument(
+ "-b", "--baudrate", default=115200, help="the baud rate of the serial device"
+ )
+ cmd_parser.add_argument("-u", "--user", default="micro", help="the telnet login username")
+ cmd_parser.add_argument("-p", "--password", default="python", help="the telnet login password")
+ cmd_parser.add_argument("-c", "--command", help="program passed in as string")
+ cmd_parser.add_argument(
+ "-w",
+ "--wait",
+ default=0,
+ type=int,
+ help="seconds to wait for USB connected board to become available",
+ )
group = cmd_parser.add_mutually_exclusive_group()
- group.add_argument('--follow', action='store_true', help='follow the output after running the scripts [default if no scripts given]')
- group.add_argument('--no-follow', action='store_true', help='Do not follow the output after running the scripts.')
- cmd_parser.add_argument('-f', '--filesystem', action='store_true', help='perform a filesystem action')
- cmd_parser.add_argument('files', nargs='*', help='input files')
+ group.add_argument(
+ "--follow",
+ action="store_true",
+ help="follow the output after running the scripts [default if no scripts given]",
+ )
+ group.add_argument(
+ "--no-follow",
+ action="store_true",
+ help="Do not follow the output after running the scripts.",
+ )
+ cmd_parser.add_argument(
+ "-f", "--filesystem", action="store_true", help="perform a filesystem action"
+ )
+ cmd_parser.add_argument("files", nargs="*", help="input files")
args = cmd_parser.parse_args()
# open the connection to the pyboard
@@ -551,7 +615,9 @@ def main():
pyb.exec_raw_no_follow(buf)
ret_err = None
else:
- ret, ret_err = pyb.exec_raw(buf, timeout=None, data_consumer=stdout_write_bytes)
+ ret, ret_err = pyb.exec_raw(
+ buf, timeout=None, data_consumer=stdout_write_bytes
+ )
except PyboardError as er:
print(er)
pyb.close()
@@ -571,14 +637,14 @@ def main():
# run the command, if given
if args.command is not None:
- execbuffer(args.command.encode('utf-8'))
+ execbuffer(args.command.encode("utf-8"))
# run any files
for filename in args.files:
- with open(filename, 'rb') as f:
+ with open(filename, "rb") as f:
pyfile = f.read()
- if filename.endswith('.mpy') and pyfile[0] == ord('M'):
- pyb.exec_('_injected_buf=' + repr(pyfile))
+ if filename.endswith(".mpy") and pyfile[0] == ord("M"):
+ pyb.exec_("_injected_buf=" + repr(pyfile))
pyfile = _injected_import_hook_code
execbuffer(pyfile)
@@ -602,5 +668,6 @@ def main():
# close the connection to the pyboard
pyb.close()
+
if __name__ == "__main__":
main()
diff --git a/tools/pydfu.py b/tools/pydfu.py
index 962ba2b7f..9f345d016 100755
--- a/tools/pydfu.py
+++ b/tools/pydfu.py
@@ -25,34 +25,34 @@ import zlib
# VID/PID
__VID = 0x0483
-__PID = 0xdf11
+__PID = 0xDF11
# USB request __TIMEOUT
__TIMEOUT = 4000
# DFU commands
-__DFU_DETACH = 0
-__DFU_DNLOAD = 1
-__DFU_UPLOAD = 2
+__DFU_DETACH = 0
+__DFU_DNLOAD = 1
+__DFU_UPLOAD = 2
__DFU_GETSTATUS = 3
__DFU_CLRSTATUS = 4
-__DFU_GETSTATE = 5
-__DFU_ABORT = 6
+__DFU_GETSTATE = 5
+__DFU_ABORT = 6
# DFU status
-__DFU_STATE_APP_IDLE = 0x00
-__DFU_STATE_APP_DETACH = 0x01
-__DFU_STATE_DFU_IDLE = 0x02
-__DFU_STATE_DFU_DOWNLOAD_SYNC = 0x03
-__DFU_STATE_DFU_DOWNLOAD_BUSY = 0x04
-__DFU_STATE_DFU_DOWNLOAD_IDLE = 0x05
-__DFU_STATE_DFU_MANIFEST_SYNC = 0x06
-__DFU_STATE_DFU_MANIFEST = 0x07
-__DFU_STATE_DFU_MANIFEST_WAIT_RESET = 0x08
-__DFU_STATE_DFU_UPLOAD_IDLE = 0x09
-__DFU_STATE_DFU_ERROR = 0x0a
+__DFU_STATE_APP_IDLE = 0x00
+__DFU_STATE_APP_DETACH = 0x01
+__DFU_STATE_DFU_IDLE = 0x02
+__DFU_STATE_DFU_DOWNLOAD_SYNC = 0x03
+__DFU_STATE_DFU_DOWNLOAD_BUSY = 0x04
+__DFU_STATE_DFU_DOWNLOAD_IDLE = 0x05
+__DFU_STATE_DFU_MANIFEST_SYNC = 0x06
+__DFU_STATE_DFU_MANIFEST = 0x07
+__DFU_STATE_DFU_MANIFEST_WAIT_RESET = 0x08
+__DFU_STATE_DFU_UPLOAD_IDLE = 0x09
+__DFU_STATE_DFU_ERROR = 0x0A
-_DFU_DESCRIPTOR_TYPE = 0x21
+_DFU_DESCRIPTOR_TYPE = 0x21
# USB device handle
@@ -68,12 +68,14 @@ __DFU_INTERFACE = 0
# Python 3 deprecated getargspec in favour of getfullargspec, but
# Python 2 doesn't have the latter, so detect which one to use
-getargspec = getattr(inspect, 'getfullargspec', inspect.getargspec)
+getargspec = getattr(inspect, "getfullargspec", inspect.getargspec)
-if 'length' in getargspec(usb.util.get_string).args:
+if "length" in getargspec(usb.util.get_string).args:
# PyUSB 1.0.0.b1 has the length argument
def get_string(dev, index):
return usb.util.get_string(dev, 255, index)
+
+
else:
# PyUSB 1.0.0.b2 dropped the length argument
def get_string(dev, index):
@@ -83,17 +85,17 @@ else:
def find_dfu_cfg_descr(descr):
if len(descr) == 9 and descr[0] == 9 and descr[1] == _DFU_DESCRIPTOR_TYPE:
nt = collections.namedtuple(
- 'CfgDescr',
+ "CfgDescr",
[
- 'bLength',
- 'bDescriptorType',
- 'bmAttributes',
- 'wDetachTimeOut',
- 'wTransferSize',
- 'bcdDFUVersion'
- ]
+ "bLength",
+ "bDescriptorType",
+ "bmAttributes",
+ "wDetachTimeOut",
+ "wTransferSize",
+ "bcdDFUVersion",
+ ],
)
- return nt(*struct.unpack('<BBBHHH', bytearray(descr)))
+ return nt(*struct.unpack("<BBBHHH", bytearray(descr)))
return None
@@ -102,9 +104,9 @@ def init():
global __dev, __cfg_descr
devices = get_dfu_devices(idVendor=__VID, idProduct=__PID)
if not devices:
- raise ValueError('No DFU device found')
+ raise ValueError("No DFU device found")
if len(devices) > 1:
- raise ValueError('Multiple DFU devices found')
+ raise ValueError("Multiple DFU devices found")
__dev = devices[0]
__dev.set_configuration()
@@ -127,8 +129,7 @@ def init():
status = get_status()
if status == __DFU_STATE_DFU_IDLE:
break
- elif (status == __DFU_STATE_DFU_DOWNLOAD_IDLE
- or status == __DFU_STATE_DFU_UPLOAD_IDLE):
+ elif status == __DFU_STATE_DFU_DOWNLOAD_IDLE or status == __DFU_STATE_DFU_UPLOAD_IDLE:
abort_request()
else:
clr_status()
@@ -141,64 +142,61 @@ def abort_request():
def clr_status():
"""Clears any error status (perhaps left over from a previous session)."""
- __dev.ctrl_transfer(0x21, __DFU_CLRSTATUS, 0, __DFU_INTERFACE,
- None, __TIMEOUT)
+ __dev.ctrl_transfer(0x21, __DFU_CLRSTATUS, 0, __DFU_INTERFACE, None, __TIMEOUT)
def get_status():
"""Get the status of the last operation."""
- stat = __dev.ctrl_transfer(0xA1, __DFU_GETSTATUS, 0, __DFU_INTERFACE,
- 6, 20000)
+ stat = __dev.ctrl_transfer(0xA1, __DFU_GETSTATUS, 0, __DFU_INTERFACE, 6, 20000)
return stat[4]
def mass_erase():
"""Performs a MASS erase (i.e. erases the entire device)."""
# Send DNLOAD with first byte=0x41
- __dev.ctrl_transfer(0x21, __DFU_DNLOAD, 0, __DFU_INTERFACE,
- '\x41', __TIMEOUT)
+ __dev.ctrl_transfer(0x21, __DFU_DNLOAD, 0, __DFU_INTERFACE, "\x41", __TIMEOUT)
# Execute last command
if get_status() != __DFU_STATE_DFU_DOWNLOAD_BUSY:
- raise Exception('DFU: erase failed')
+ raise Exception("DFU: erase failed")
# Check command state
if get_status() != __DFU_STATE_DFU_DOWNLOAD_IDLE:
- raise Exception('DFU: erase failed')
+ raise Exception("DFU: erase failed")
def page_erase(addr):
"""Erases a single page."""
if __verbose:
- print('Erasing page: 0x%x...' % (addr))
+ print("Erasing page: 0x%x..." % (addr))
# Send DNLOAD with first byte=0x41 and page address
- buf = struct.pack('<BI', 0x41, addr)
+ buf = struct.pack("<BI", 0x41, addr)
__dev.ctrl_transfer(0x21, __DFU_DNLOAD, 0, __DFU_INTERFACE, buf, __TIMEOUT)
# Execute last command
if get_status() != __DFU_STATE_DFU_DOWNLOAD_BUSY:
- raise Exception('DFU: erase failed')
+ raise Exception("DFU: erase failed")
# Check command state
if get_status() != __DFU_STATE_DFU_DOWNLOAD_IDLE:
- raise Exception('DFU: erase failed')
+ raise Exception("DFU: erase failed")
def set_address(addr):
"""Sets the address for the next operation."""
# Send DNLOAD with first byte=0x21 and page address
- buf = struct.pack('<BI', 0x21, addr)
+ buf = struct.pack("<BI", 0x21, addr)
__dev.ctrl_transfer(0x21, __DFU_DNLOAD, 0, __DFU_INTERFACE, buf, __TIMEOUT)
# Execute last command
if get_status() != __DFU_STATE_DFU_DOWNLOAD_BUSY:
- raise Exception('DFU: set address failed')
+ raise Exception("DFU: set address failed")
# Check command state
if get_status() != __DFU_STATE_DFU_DOWNLOAD_IDLE:
- raise Exception('DFU: set address failed')
+ raise Exception("DFU: set address failed")
def write_memory(addr, buf, progress=None, progress_addr=0, progress_size=0):
@@ -213,28 +211,29 @@ def write_memory(addr, buf, progress=None, progress_addr=0, progress_size=0):
while xfer_bytes < xfer_total:
if __verbose and xfer_count % 512 == 0:
- print('Addr 0x%x %dKBs/%dKBs...' % (xfer_base + xfer_bytes,
- xfer_bytes // 1024,
- xfer_total // 1024))
+ print(
+ "Addr 0x%x %dKBs/%dKBs..."
+ % (xfer_base + xfer_bytes, xfer_bytes // 1024, xfer_total // 1024)
+ )
if progress and xfer_count % 2 == 0:
- progress(progress_addr, xfer_base + xfer_bytes - progress_addr,
- progress_size)
+ progress(progress_addr, xfer_base + xfer_bytes - progress_addr, progress_size)
# Set mem write address
set_address(xfer_base + xfer_bytes)
# Send DNLOAD with fw data
chunk = min(__cfg_descr.wTransferSize, xfer_total - xfer_bytes)
- __dev.ctrl_transfer(0x21, __DFU_DNLOAD, 2, __DFU_INTERFACE,
- buf[xfer_bytes:xfer_bytes + chunk], __TIMEOUT)
+ __dev.ctrl_transfer(
+ 0x21, __DFU_DNLOAD, 2, __DFU_INTERFACE, buf[xfer_bytes : xfer_bytes + chunk], __TIMEOUT
+ )
# Execute last command
if get_status() != __DFU_STATE_DFU_DOWNLOAD_BUSY:
- raise Exception('DFU: write memory failed')
+ raise Exception("DFU: write memory failed")
# Check command state
if get_status() != __DFU_STATE_DFU_DOWNLOAD_IDLE:
- raise Exception('DFU: write memory failed')
+ raise Exception("DFU: write memory failed")
xfer_count += 1
xfer_bytes += chunk
@@ -255,14 +254,14 @@ def write_page(buf, xfer_offset):
# Execute last command
if get_status() != __DFU_STATE_DFU_DOWNLOAD_BUSY:
- raise Exception('DFU: write memory failed')
+ raise Exception("DFU: write memory failed")
# Check command state
if get_status() != __DFU_STATE_DFU_DOWNLOAD_IDLE:
- raise Exception('DFU: write memory failed')
+ raise Exception("DFU: write memory failed")
if __verbose:
- print('Write: 0x%x ' % (xfer_base + xfer_offset))
+ print("Write: 0x%x " % (xfer_base + xfer_offset))
def exit_dfu():
@@ -271,13 +270,12 @@ def exit_dfu():
set_address(0x08000000)
# Send DNLOAD with 0 length to exit DFU
- __dev.ctrl_transfer(0x21, __DFU_DNLOAD, 0, __DFU_INTERFACE,
- None, __TIMEOUT)
+ __dev.ctrl_transfer(0x21, __DFU_DNLOAD, 0, __DFU_INTERFACE, None, __TIMEOUT)
try:
# Execute last command
if get_status() != __DFU_STATE_DFU_MANIFEST:
- print('Failed to reset device')
+ print("Failed to reset device")
# Release device
usb.util.dispose_resources(__dev)
@@ -301,7 +299,7 @@ def consume(fmt, data, names):
def cstring(string):
"""Extracts a null-terminated string from a byte array."""
- return string.decode('utf-8').split('\0', 1)[0]
+ return string.decode("utf-8").split("\0", 1)[0]
def compute_crc(data):
@@ -320,8 +318,8 @@ def read_dfu_file(filename):
If an error occurs while parsing the file, then None is returned.
"""
- print('File: {}'.format(filename))
- with open(filename, 'rb') as fin:
+ print("File: {}".format(filename))
+ with open(filename, "rb") as fin:
data = fin.read()
crc = compute_crc(data[:-4])
elements = []
@@ -334,11 +332,12 @@ def read_dfu_file(filename):
# B uint8_t version 1
# I uint32_t size Size of the DFU file (without suffix)
# B uint8_t targets Number of targets
- dfu_prefix, data = consume('<5sBIB', data,
- 'signature version size targets')
- print(' %(signature)s v%(version)d, image size: %(size)d, '
- 'targets: %(targets)d' % dfu_prefix)
- for target_idx in range(dfu_prefix['targets']):
+ dfu_prefix, data = consume("<5sBIB", data, "signature version size targets")
+ print(
+ " %(signature)s v%(version)d, image size: %(size)d, "
+ "targets: %(targets)d" % dfu_prefix
+ )
+ for target_idx in range(dfu_prefix["targets"]):
# Decode the Image Prefix
#
# <6sBI255s2I
@@ -349,40 +348,40 @@ def read_dfu_file(filename):
# 255s char[255] name Name of the target
# I uint32_t size Size of image (without prefix)
# I uint32_t elements Number of elements in the image
- img_prefix, data = consume('<6sBI255s2I', data,
- 'signature altsetting named name '
- 'size elements')
- img_prefix['num'] = target_idx
- if img_prefix['named']:
- img_prefix['name'] = cstring(img_prefix['name'])
+ img_prefix, data = consume(
+ "<6sBI255s2I", data, "signature altsetting named name " "size elements"
+ )
+ img_prefix["num"] = target_idx
+ if img_prefix["named"]:
+ img_prefix["name"] = cstring(img_prefix["name"])
else:
- img_prefix['name'] = ''
- print(' %(signature)s %(num)d, alt setting: %(altsetting)s, '
- 'name: "%(name)s", size: %(size)d, elements: %(elements)d'
- % img_prefix)
+ img_prefix["name"] = ""
+ print(
+ " %(signature)s %(num)d, alt setting: %(altsetting)s, "
+ 'name: "%(name)s", size: %(size)d, elements: %(elements)d' % img_prefix
+ )
- target_size = img_prefix['size']
+ target_size = img_prefix["size"]
target_data = data[:target_size]
data = data[target_size:]
- for elem_idx in range(img_prefix['elements']):
+ for elem_idx in range(img_prefix["elements"]):
# Decode target prefix
#
# <2I
# < little endian Endianness
# I uint32_t element Address
# I uint32_t element Size
- elem_prefix, target_data = consume('<2I', target_data, 'addr size')
- elem_prefix['num'] = elem_idx
- print(' %(num)d, address: 0x%(addr)08x, size: %(size)d'
- % elem_prefix)
- elem_size = elem_prefix['size']
+ elem_prefix, target_data = consume("<2I", target_data, "addr size")
+ elem_prefix["num"] = elem_idx
+ print(" %(num)d, address: 0x%(addr)08x, size: %(size)d" % elem_prefix)
+ elem_size = elem_prefix["size"]
elem_data = target_data[:elem_size]
target_data = target_data[elem_size:]
- elem_prefix['data'] = elem_data
+ elem_prefix["data"] = elem_data
elements.append(elem_prefix)
if len(target_data):
- print('target %d PARSE ERROR' % target_idx)
+ print("target %d PARSE ERROR" % target_idx)
# Decode DFU Suffix
#
@@ -395,16 +394,19 @@ def read_dfu_file(filename):
# 3s char[3] ufd "UFD"
# B uint8_t len 16
# I uint32_t crc32 Checksum
- dfu_suffix = named(struct.unpack('<4H3sBI', data[:16]),
- 'device product vendor dfu ufd len crc')
- print(' usb: %(vendor)04x:%(product)04x, device: 0x%(device)04x, '
- 'dfu: 0x%(dfu)04x, %(ufd)s, %(len)d, 0x%(crc)08x' % dfu_suffix)
- if crc != dfu_suffix['crc']:
- print('CRC ERROR: computed crc32 is 0x%08x' % crc)
+ dfu_suffix = named(
+ struct.unpack("<4H3sBI", data[:16]), "device product vendor dfu ufd len crc"
+ )
+ print(
+ " usb: %(vendor)04x:%(product)04x, device: 0x%(device)04x, "
+ "dfu: 0x%(dfu)04x, %(ufd)s, %(len)d, 0x%(crc)08x" % dfu_suffix
+ )
+ if crc != dfu_suffix["crc"]:
+ print("CRC ERROR: computed crc32 is 0x%08x" % crc)
return
data = data[16:]
if data:
- print('PARSE ERROR')
+ print("PARSE ERROR")
return
return elements
@@ -418,8 +420,7 @@ class FilterDFU(object):
def __call__(self, device):
for cfg in device:
for intf in cfg:
- return (intf.bInterfaceClass == 0xFE and
- intf.bInterfaceSubClass == 1)
+ return intf.bInterfaceClass == 0xFE and intf.bInterfaceSubClass == 1
def get_dfu_devices(*args, **kwargs):
@@ -429,8 +430,7 @@ def get_dfu_devices(*args, **kwargs):
"""
# Convert to list for compatibility with newer PyUSB
- return list(usb.core.find(*args, find_all=True,
- custom_match=FilterDFU(), **kwargs))
+ return list(usb.core.find(*args, find_all=True, custom_match=FilterDFU(), **kwargs))
def get_memory_layout(device):
@@ -446,25 +446,29 @@ def get_memory_layout(device):
cfg = device[0]
intf = cfg[(0, 0)]
mem_layout_str = get_string(device, intf.iInterface)
- mem_layout = mem_layout_str.split('/')
+ mem_layout = mem_layout_str.split("/")
result = []
for mem_layout_index in range(1, len(mem_layout), 2):
addr = int(mem_layout[mem_layout_index], 0)
- segments = mem_layout[mem_layout_index + 1].split(',')
- seg_re = re.compile(r'(\d+)\*(\d+)(.)(.)')
+ segments = mem_layout[mem_layout_index + 1].split(",")
+ seg_re = re.compile(r"(\d+)\*(\d+)(.)(.)")
for segment in segments:
seg_match = seg_re.match(segment)
num_pages = int(seg_match.groups()[0], 10)
page_size = int(seg_match.groups()[1], 10)
multiplier = seg_match.groups()[2]
- if multiplier == 'K':
+ if multiplier == "K":
page_size *= 1024
- if multiplier == 'M':
+ if multiplier == "M":
page_size *= 1024 * 1024
size = num_pages * page_size
last_addr = addr + size - 1
- result.append(named((addr, last_addr, size, num_pages, page_size),
- 'addr last_addr size num_pages page_size'))
+ result.append(
+ named(
+ (addr, last_addr, size, num_pages, page_size),
+ "addr last_addr size num_pages page_size",
+ )
+ )
addr += size
return result
@@ -473,18 +477,22 @@ def list_dfu_devices(*args, **kwargs):
"""Prints a lits of devices detected in DFU mode."""
devices = get_dfu_devices(*args, **kwargs)
if not devices:
- print('No DFU capable devices found')
+ print("No DFU capable devices found")
return
for device in devices:
- print('Bus {} Device {:03d}: ID {:04x}:{:04x}'
- .format(device.bus, device.address,
- device.idVendor, device.idProduct))
+ print(
+ "Bus {} Device {:03d}: ID {:04x}:{:04x}".format(
+ device.bus, device.address, device.idVendor, device.idProduct
+ )
+ )
layout = get_memory_layout(device)
- print('Memory Layout')
+ print("Memory Layout")
for entry in layout:
- print(' 0x{:x} {:2d} pages of {:3d}K bytes'
- .format(entry['addr'], entry['num_pages'],
- entry['page_size'] // 1024))
+ print(
+ " 0x{:x} {:2d} pages of {:3d}K bytes".format(
+ entry["addr"], entry["num_pages"], entry["page_size"] // 1024
+ )
+ )
def write_elements(elements, mass_erase_used, progress=None):
@@ -494,9 +502,9 @@ def write_elements(elements, mass_erase_used, progress=None):
mem_layout = get_memory_layout(__dev)
for elem in elements:
- addr = elem['addr']
- size = elem['size']
- data = elem['data']
+ addr = elem["addr"]
+ size = elem["size"]
+ data = elem["data"]
elem_size = size
elem_addr = addr
if progress:
@@ -505,18 +513,16 @@ def write_elements(elements, mass_erase_used, progress=None):
write_size = size
if not mass_erase_used:
for segment in mem_layout:
- if addr >= segment['addr'] and \
- addr <= segment['last_addr']:
+ if addr >= segment["addr"] and addr <= segment["last_addr"]:
# We found the page containing the address we want to
# write, erase it
- page_size = segment['page_size']
+ page_size = segment["page_size"]
page_addr = addr & ~(page_size - 1)
if addr + write_size > page_addr + page_size:
write_size = page_addr + page_size - addr
page_erase(page_addr)
break
- write_memory(addr, data[:write_size], progress,
- elem_addr, elem_size)
+ write_memory(addr, data[:write_size], progress, elem_addr, elem_size)
data = data[write_size:]
addr += write_size
size -= write_size
@@ -528,45 +534,36 @@ def cli_progress(addr, offset, size):
"""Prints a progress report suitable for use on the command line."""
width = 25
done = offset * width // size
- print('\r0x{:08x} {:7d} [{}{}] {:3d}% '
- .format(addr, size, '=' * done, ' ' * (width - done),
- offset * 100 // size), end='')
+ print(
+ "\r0x{:08x} {:7d} [{}{}] {:3d}% ".format(
+ addr, size, "=" * done, " " * (width - done), offset * 100 // size
+ ),
+ end="",
+ )
try:
sys.stdout.flush()
except OSError:
- pass # Ignore Windows CLI "WinError 87" on Python 3.6
+ pass # Ignore Windows CLI "WinError 87" on Python 3.6
if offset == size:
- print('')
+ print("")
def main():
"""Test program for verifying this files functionality."""
global __verbose
# Parse CMD args
- parser = argparse.ArgumentParser(description='DFU Python Util')
+ parser = argparse.ArgumentParser(description="DFU Python Util")
parser.add_argument(
- '-l', '--list',
- help='list available DFU devices',
- action='store_true',
- default=False
+ "-l", "--list", help="list available DFU devices", action="store_true", default=False
)
parser.add_argument(
- '-m', '--mass-erase',
- help='mass erase device',
- action='store_true',
- default=False
+ "-m", "--mass-erase", help="mass erase device", action="store_true", default=False
)
parser.add_argument(
- '-u', '--upload',
- help='read file from DFU device',
- dest='path',
- default=False
+ "-u", "--upload", help="read file from DFU device", dest="path", default=False
)
parser.add_argument(
- '-v', '--verbose',
- help='increase output verbosity',
- action='store_true',
- default=False
+ "-v", "--verbose", help="increase output verbosity", action="store_true", default=False
)
args = parser.parse_args()
@@ -579,21 +576,22 @@ def main():
init()
if args.mass_erase:
- print('Mass erase...')
+ print("Mass erase...")
mass_erase()
if args.path:
elements = read_dfu_file(args.path)
if not elements:
return
- print('Writing memory...')
+ print("Writing memory...")
write_elements(elements, args.mass_erase, progress=cli_progress)
- print('Exiting DFU...')
+ print("Exiting DFU...")
exit_dfu()
return
- print('No command specified')
+ print("No command specified")
+
-if __name__ == '__main__':
+if __name__ == "__main__":
main()
diff --git a/tools/tinytest-codegen.py b/tools/tinytest-codegen.py
index 424f70a9f..64f88edb3 100755
--- a/tools/tinytest-codegen.py
+++ b/tools/tinytest-codegen.py
@@ -9,17 +9,19 @@ import argparse
def escape(s):
s = s.decode()
lookup = {
- '\0': '\\0',
- '\t': '\\t',
- '\n': '\\n\"\n\"',
- '\r': '\\r',
- '\\': '\\\\',
- '\"': '\\\"',
+ "\0": "\\0",
+ "\t": "\\t",
+ "\n": '\\n"\n"',
+ "\r": "\\r",
+ "\\": "\\\\",
+ '"': '\\"',
}
- return "\"\"\n\"{}\"".format(''.join([lookup[x] if x in lookup else x for x in s]))
+ return '""\n"{}"'.format("".join([lookup[x] if x in lookup else x for x in s]))
+
def chew_filename(t):
- return { 'func': "test_{}_fn".format(sub(r'/|\.|-', '_', t)), 'desc': t }
+ return {"func": "test_{}_fn".format(sub(r"/|\.|-", "_", t)), "desc": t}
+
def script_to_map(test_file):
r = {"name": chew_filename(test_file)["func"]}
@@ -29,6 +31,7 @@ def script_to_map(test_file):
r["output"] = escape(f.read())
return r
+
test_function = (
"void {name}(void* data) {{\n"
" static const char pystr[] = {script};\n"
@@ -40,79 +43,86 @@ test_function = (
"}}"
)
-testcase_struct = (
- "struct testcase_t {name}_tests[] = {{\n{body}\n END_OF_TESTCASES\n}};"
-)
-testcase_member = (
- " {{ \"{desc}\", {func}, TT_ENABLED_, 0, 0 }},"
-)
+testcase_struct = "struct testcase_t {name}_tests[] = {{\n{body}\n END_OF_TESTCASES\n}};"
+testcase_member = ' {{ "{desc}", {func}, TT_ENABLED_, 0, 0 }},'
-testgroup_struct = (
- "struct testgroup_t groups[] = {{\n{body}\n END_OF_GROUPS\n}};"
-)
-testgroup_member = (
- " {{ \"{name}\", {name}_tests }},"
-)
+testgroup_struct = "struct testgroup_t groups[] = {{\n{body}\n END_OF_GROUPS\n}};"
+testgroup_member = ' {{ "{name}", {name}_tests }},'
## XXX: may be we could have `--without <groups>` argument...
# currently these tests are selected because they pass on qemu-arm
-test_dirs = ('basics', 'micropython', 'misc', 'extmod', 'float', 'inlineasm', 'qemu-arm',) # 'import', 'io',)
+test_dirs = (
+ "basics",
+ "micropython",
+ "misc",
+ "extmod",
+ "float",
+ "inlineasm",
+ "qemu-arm",
+) # 'import', 'io',)
exclude_tests = (
# pattern matching in .exp
- 'basics/bytes_compare3.py',
- 'extmod/ticks_diff.py',
- 'extmod/time_ms_us.py',
- 'extmod/uheapq_timeq.py',
+ "basics/bytes_compare3.py",
+ "extmod/ticks_diff.py",
+ "extmod/time_ms_us.py",
+ "extmod/uheapq_timeq.py",
# unicode char issue
- 'extmod/ujson_loads.py',
+ "extmod/ujson_loads.py",
# doesn't output to python stdout
- 'extmod/ure_debug.py',
- 'extmod/vfs_basic.py',
- 'extmod/vfs_fat_ramdisk.py', 'extmod/vfs_fat_fileio.py',
- 'extmod/vfs_fat_fsusermount.py', 'extmod/vfs_fat_oldproto.py',
+ "extmod/ure_debug.py",
+ "extmod/vfs_basic.py",
+ "extmod/vfs_fat_ramdisk.py",
+ "extmod/vfs_fat_fileio.py",
+ "extmod/vfs_fat_fsusermount.py",
+ "extmod/vfs_fat_oldproto.py",
# rounding issues
- 'float/float_divmod.py',
+ "float/float_divmod.py",
# requires double precision floating point to work
- 'float/float2int_doubleprec_intbig.py',
- 'float/float_parse_doubleprec.py',
+ "float/float2int_doubleprec_intbig.py",
+ "float/float_parse_doubleprec.py",
# inline asm FP tests (require Cortex-M4)
- 'inlineasm/asmfpaddsub.py', 'inlineasm/asmfpcmp.py', 'inlineasm/asmfpldrstr.py',
- 'inlineasm/asmfpmuldiv.py','inlineasm/asmfpsqrt.py',
+ "inlineasm/asmfpaddsub.py",
+ "inlineasm/asmfpcmp.py",
+ "inlineasm/asmfpldrstr.py",
+ "inlineasm/asmfpmuldiv.py",
+ "inlineasm/asmfpsqrt.py",
# different filename in output
- 'micropython/emg_exc.py',
- 'micropython/heapalloc_traceback.py',
+ "micropython/emg_exc.py",
+ "micropython/heapalloc_traceback.py",
# pattern matching in .exp
- 'micropython/meminfo.py',
+ "micropython/meminfo.py",
# needs sys stdfiles
- 'misc/print_exception.py',
+ "misc/print_exception.py",
# settrace .exp files are too large
- 'misc/sys_settrace_loop.py',
- 'misc/sys_settrace_generator.py',
- 'misc/sys_settrace_features.py',
+ "misc/sys_settrace_loop.py",
+ "misc/sys_settrace_generator.py",
+ "misc/sys_settrace_features.py",
)
output = []
tests = []
-argparser = argparse.ArgumentParser(description='Convert native MicroPython tests to tinytest/upytesthelper C code')
-argparser.add_argument('--stdin', action="store_true", help='read list of tests from stdin')
+argparser = argparse.ArgumentParser(
+ description="Convert native MicroPython tests to tinytest/upytesthelper C code"
+)
+argparser.add_argument("--stdin", action="store_true", help="read list of tests from stdin")
args = argparser.parse_args()
if not args.stdin:
for group in test_dirs:
- tests += [test for test in glob('{}/*.py'.format(group)) if test not in exclude_tests]
+ tests += [test for test in glob("{}/*.py".format(group)) if test not in exclude_tests]
else:
for l in sys.stdin:
tests.append(l.rstrip())
output.extend([test_function.format(**script_to_map(test)) for test in tests])
testcase_members = [testcase_member.format(**chew_filename(test)) for test in tests]
-output.append(testcase_struct.format(name="", body='\n'.join(testcase_members)))
+output.append(testcase_struct.format(name="", body="\n".join(testcase_members)))
testgroup_members = [testgroup_member.format(name=group) for group in [""]]
-output.append(testgroup_struct.format(body='\n'.join(testgroup_members)))
+output.append(testgroup_struct.format(body="\n".join(testgroup_members)))
## XXX: may be we could have `--output <filename>` argument...
# Don't depend on what system locale is set, use utf8 encoding.
-sys.stdout.buffer.write('\n\n'.join(output).encode('utf8'))
+sys.stdout.buffer.write("\n\n".join(output).encode("utf8"))
diff --git a/tools/uf2conv.py b/tools/uf2conv.py
index 2f2812abf..d67a55224 100755
--- a/tools/uf2conv.py
+++ b/tools/uf2conv.py
@@ -1,23 +1,23 @@
#!/usr/bin/env python3
# Microsoft UF2
-#
+#
# The MIT License (MIT)
-#
+#
# Copyright (c) Microsoft Corporation
-#
+#
# All rights reserved.
-#
+#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
-#
+#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
-#
+#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
@@ -35,17 +35,17 @@ import os.path
import argparse
-UF2_MAGIC_START0 = 0x0A324655 # "UF2\n"
-UF2_MAGIC_START1 = 0x9E5D5157 # Randomly selected
-UF2_MAGIC_END = 0x0AB16F30 # Ditto
+UF2_MAGIC_START0 = 0x0A324655 # "UF2\n"
+UF2_MAGIC_START1 = 0x9E5D5157 # Randomly selected
+UF2_MAGIC_END = 0x0AB16F30 # Ditto
families = {
- 'SAMD21': 0x68ed2b88,
- 'SAMD51': 0x55114460,
- 'NRF52': 0x1b57745f,
- 'STM32F1': 0x5ee21072,
- 'STM32F4': 0x57755a57,
- 'ATMEGA32': 0x16573617,
+ "SAMD21": 0x68ED2B88,
+ "SAMD51": 0x55114460,
+ "NRF52": 0x1B57745F,
+ "STM32F1": 0x5EE21072,
+ "STM32F4": 0x57755A57,
+ "ATMEGA32": 0x16573617,
}
INFO_FILE = "/INFO_UF2.TXT"
@@ -58,15 +58,17 @@ def is_uf2(buf):
w = struct.unpack("<II", buf[0:8])
return w[0] == UF2_MAGIC_START0 and w[1] == UF2_MAGIC_START1
+
def is_hex(buf):
try:
w = buf[0:30].decode("utf-8")
except UnicodeDecodeError:
return False
- if w[0] == ':' and re.match(b"^[:0-9a-fA-F\r\n]+$", buf):
+ if w[0] == ":" and re.match(b"^[:0-9a-fA-F\r\n]+$", buf):
return True
return False
+
def convert_from_uf2(buf):
global appstartaddr
numblocks = len(buf) // 512
@@ -74,7 +76,7 @@ def convert_from_uf2(buf):
outp = b""
for blockno in range(numblocks):
ptr = blockno * 512
- block = buf[ptr:ptr + 512]
+ block = buf[ptr : ptr + 512]
hd = struct.unpack(b"<IIIIIIII", block[0:32])
if hd[0] != UF2_MAGIC_START0 or hd[1] != UF2_MAGIC_START1:
print("Skipping block at " + ptr + "; bad magic")
@@ -92,7 +94,7 @@ def convert_from_uf2(buf):
padding = newaddr - curraddr
if padding < 0:
assert False, "Block out of order at " + ptr
- if padding > 10*1024*1024:
+ if padding > 10 * 1024 * 1024:
assert False, "More than 10M of padding needed at " + ptr
if padding % 4 != 0:
assert False, "Non-word padding size at " + ptr
@@ -103,6 +105,7 @@ def convert_from_uf2(buf):
curraddr = newaddr + datalen
return outp
+
def convert_to_carray(file_content):
outp = "const unsigned char bindata[] __attribute__((aligned(16))) = {"
for i in range(len(file_content)):
@@ -112,6 +115,7 @@ def convert_to_carray(file_content):
outp += "\n};\n"
return outp
+
def convert_to_uf2(file_content):
global familyid
datapadding = b""
@@ -121,13 +125,21 @@ def convert_to_uf2(file_content):
outp = b""
for blockno in range(numblocks):
ptr = 256 * blockno
- chunk = file_content[ptr:ptr + 256]
+ chunk = file_content[ptr : ptr + 256]
flags = 0x0
if familyid:
flags |= 0x2000
- hd = struct.pack(b"<IIIIIIII",
- UF2_MAGIC_START0, UF2_MAGIC_START1,
- flags, ptr + appstartaddr, 256, blockno, numblocks, familyid)
+ hd = struct.pack(
+ b"<IIIIIIII",
+ UF2_MAGIC_START0,
+ UF2_MAGIC_START1,
+ flags,
+ ptr + appstartaddr,
+ 256,
+ blockno,
+ numblocks,
+ familyid,
+ )
while len(chunk) < 256:
chunk += b"\x00"
block = hd + chunk + datapadding + struct.pack(b"<I", UF2_MAGIC_END)
@@ -135,6 +147,7 @@ def convert_to_uf2(file_content):
outp += block
return outp
+
class Block:
def __init__(self, addr):
self.addr = addr
@@ -145,35 +158,44 @@ class Block:
flags = 0x0
if familyid:
flags |= 0x2000
- hd = struct.pack("<IIIIIIII",
- UF2_MAGIC_START0, UF2_MAGIC_START1,
- flags, self.addr, 256, blockno, numblocks, familyid)
+ hd = struct.pack(
+ "<IIIIIIII",
+ UF2_MAGIC_START0,
+ UF2_MAGIC_START1,
+ flags,
+ self.addr,
+ 256,
+ blockno,
+ numblocks,
+ familyid,
+ )
hd += self.bytes[0:256]
while len(hd) < 512 - 4:
hd += b"\x00"
hd += struct.pack("<I", UF2_MAGIC_END)
return hd
+
def convert_from_hex_to_uf2(buf):
global appstartaddr
appstartaddr = None
upper = 0
currblock = None
blocks = []
- for line in buf.split('\n'):
+ for line in buf.split("\n"):
if line[0] != ":":
continue
i = 1
rec = []
while i < len(line) - 1:
- rec.append(int(line[i:i+2], 16))
+ rec.append(int(line[i : i + 2], 16))
i += 2
tp = rec[3]
if tp == 4:
upper = ((rec[4] << 8) | rec[5]) << 16
elif tp == 2:
upper = ((rec[4] << 8) | rec[5]) << 4
- assert (upper & 0xffff) == 0
+ assert (upper & 0xFFFF) == 0
elif tp == 1:
break
elif tp == 0:
@@ -182,10 +204,10 @@ def convert_from_hex_to_uf2(buf):
appstartaddr = addr
i = 4
while i < len(rec) - 1:
- if not currblock or currblock.addr & ~0xff != addr & ~0xff:
- currblock = Block(addr & ~0xff)
+ if not currblock or currblock.addr & ~0xFF != addr & ~0xFF:
+ currblock = Block(addr & ~0xFF)
blocks.append(currblock)
- currblock.bytes[addr & 0xff] = rec[i]
+ currblock.bytes[addr & 0xFF] = rec[i]
addr += 1
i += 1
numblocks = len(blocks)
@@ -194,14 +216,24 @@ def convert_from_hex_to_uf2(buf):
resfile += blocks[i].encode(i, numblocks)
return resfile
+
def get_drives():
drives = []
if sys.platform == "win32":
- r = subprocess.check_output(["wmic", "PATH", "Win32_LogicalDisk",
- "get", "DeviceID,", "VolumeName,",
- "FileSystem,", "DriveType"])
- for line in r.split('\n'):
- words = re.split('\s+', line)
+ r = subprocess.check_output(
+ [
+ "wmic",
+ "PATH",
+ "Win32_LogicalDisk",
+ "get",
+ "DeviceID,",
+ "VolumeName,",
+ "FileSystem,",
+ "DriveType",
+ ]
+ )
+ for line in r.split("\n"):
+ words = re.split("\s+", line)
if len(words) >= 3 and words[1] == "2" and words[2] == "FAT":
drives.append(words[0])
else:
@@ -215,7 +247,6 @@ def get_drives():
for d in os.listdir(rootpath):
drives.append(os.path.join(rootpath, d))
-
def has_info(d):
try:
return os.path.isfile(d + INFO_FILE)
@@ -226,7 +257,7 @@ def get_drives():
def board_id(path):
- with open(path + INFO_FILE, mode='r') as file:
+ with open(path + INFO_FILE, mode="r") as file:
file_content = file.read()
return re.search("Board-ID: ([^\r\n]*)", file_content).group(1)
@@ -244,28 +275,45 @@ def write_file(name, buf):
def main():
global appstartaddr, familyid
+
def error(msg):
print(msg)
sys.exit(1)
- parser = argparse.ArgumentParser(description='Convert to UF2 or flash directly.')
- parser.add_argument('input', metavar='INPUT', type=str, nargs='?',
- help='input file (HEX, BIN or UF2)')
- parser.add_argument('-b' , '--base', dest='base', type=str,
- default="0x2000",
- help='set base address of application for BIN format (default: 0x2000)')
- parser.add_argument('-o' , '--output', metavar="FILE", dest='output', type=str,
- help='write output to named file; defaults to "flash.uf2" or "flash.bin" where sensible')
- parser.add_argument('-d' , '--device', dest="device_path",
- help='select a device path to flash')
- parser.add_argument('-l' , '--list', action='store_true',
- help='list connected devices')
- parser.add_argument('-c' , '--convert', action='store_true',
- help='do not flash, just convert')
- parser.add_argument('-f' , '--family', dest='family', type=str,
- default="0x0",
- help='specify familyID - number or name (default: 0x0)')
- parser.add_argument('-C' , '--carray', action='store_true',
- help='convert binary file to a C array, not UF2')
+
+ parser = argparse.ArgumentParser(description="Convert to UF2 or flash directly.")
+ parser.add_argument(
+ "input", metavar="INPUT", type=str, nargs="?", help="input file (HEX, BIN or UF2)"
+ )
+ parser.add_argument(
+ "-b",
+ "--base",
+ dest="base",
+ type=str,
+ default="0x2000",
+ help="set base address of application for BIN format (default: 0x2000)",
+ )
+ parser.add_argument(
+ "-o",
+ "--output",
+ metavar="FILE",
+ dest="output",
+ type=str,
+ help='write output to named file; defaults to "flash.uf2" or "flash.bin" where sensible',
+ )
+ parser.add_argument("-d", "--device", dest="device_path", help="select a device path to flash")
+ parser.add_argument("-l", "--list", action="store_true", help="list connected devices")
+ parser.add_argument("-c", "--convert", action="store_true", help="do not flash, just convert")
+ parser.add_argument(
+ "-f",
+ "--family",
+ dest="family",
+ type=str,
+ default="0x0",
+ help="specify familyID - number or name (default: 0x0)",
+ )
+ parser.add_argument(
+ "-C", "--carray", action="store_true", help="convert binary file to a C array, not UF2"
+ )
args = parser.parse_args()
appstartaddr = int(args.base, 0)
@@ -282,7 +330,7 @@ def main():
else:
if not args.input:
error("Need input file")
- with open(args.input, mode='rb') as f:
+ with open(args.input, mode="rb") as f:
inpbuf = f.read()
from_uf2 = is_uf2(inpbuf)
ext = "uf2"
@@ -296,8 +344,10 @@ def main():
ext = "h"
else:
outbuf = convert_to_uf2(inpbuf)
- print("Converting to %s, output size: %d, start address: 0x%x" %
- (ext, len(outbuf), appstartaddr))
+ print(
+ "Converting to %s, output size: %d, start address: 0x%x"
+ % (ext, len(outbuf), appstartaddr)
+ )
if args.convert:
drives = []
if args.output == None:
diff --git a/tools/upip.py b/tools/upip.py
index e3b888516..0036eac25 100644
--- a/tools/upip.py
+++ b/tools/upip.py
@@ -12,6 +12,7 @@ import uerrno as errno
import ujson as json
import uzlib
import upip_utarfile as tarfile
+
gc.collect()
@@ -23,9 +24,11 @@ gzdict_sz = 16 + 15
file_buf = bytearray(512)
+
class NotFoundError(Exception):
pass
+
def op_split(path):
if path == "":
return ("", "")
@@ -37,9 +40,11 @@ def op_split(path):
head = "/"
return (head, r[1])
+
def op_basename(path):
return op_split(path)[1]
+
# Expects *file* name
def _makedirs(name, mode=0o777):
ret = False
@@ -70,26 +75,27 @@ def save_file(fname, subf):
break
outf.write(file_buf, sz)
+
def install_tar(f, prefix):
meta = {}
for info in f:
- #print(info)
+ # print(info)
fname = info.name
try:
- fname = fname[fname.index("/") + 1:]
+ fname = fname[fname.index("/") + 1 :]
except ValueError:
fname = ""
save = True
for p in ("setup.", "PKG-INFO", "README"):
- #print(fname, p)
- if fname.startswith(p) or ".egg-info" in fname:
- if fname.endswith("/requires.txt"):
- meta["deps"] = f.extractfile(info).read()
- save = False
- if debug:
- print("Skipping", fname)
- break
+ # print(fname, p)
+ if fname.startswith(p) or ".egg-info" in fname:
+ if fname.endswith("/requires.txt"):
+ meta["deps"] = f.extractfile(info).read()
+ save = False
+ if debug:
+ print("Skipping", fname)
+ break
if save:
outfname = prefix + fname
@@ -101,32 +107,37 @@ def install_tar(f, prefix):
save_file(outfname, subf)
return meta
+
def expandhome(s):
if "~/" in s:
h = os.getenv("HOME")
s = s.replace("~/", h + "/")
return s
+
import ussl
import usocket
+
warn_ussl = True
+
+
def url_open(url):
global warn_ussl
if debug:
print(url)
- proto, _, host, urlpath = url.split('/', 3)
+ proto, _, host, urlpath = url.split("/", 3)
try:
ai = usocket.getaddrinfo(host, 443, 0, usocket.SOCK_STREAM)
except OSError as e:
fatal("Unable to resolve %s (no Internet?)" % host, e)
- #print("Address infos:", ai)
+ # print("Address infos:", ai)
ai = ai[0]
s = usocket.socket(ai[0], ai[1], ai[2])
try:
- #print("Connect address:", addr)
+ # print("Connect address:", addr)
s.connect(ai[-1])
if proto == "https:":
@@ -147,7 +158,7 @@ def url_open(url):
l = s.readline()
if not l:
raise ValueError("Unexpected EOF in HTTP headers")
- if l == b'\r\n':
+ if l == b"\r\n":
break
except Exception as e:
s.close()
@@ -175,6 +186,7 @@ def fatal(msg, exc=None):
raise exc
sys.exit(1)
+
def install_pkg(pkg_spec, install_path):
data = get_pkg_metadata(pkg_spec)
@@ -198,6 +210,7 @@ def install_pkg(pkg_spec, install_path):
gc.collect()
return meta
+
def install(to_install, install_path=None):
# Calculate gzip dictionary size to use
global gzdict_sz
@@ -230,9 +243,11 @@ def install(to_install, install_path=None):
deps = deps.decode("utf-8").split("\n")
to_install.extend(deps)
except Exception as e:
- print("Error installing '{}': {}, packages may be partially installed".format(
- pkg_spec, e),
- file=sys.stderr)
+ print(
+ "Error installing '{}': {}, packages may be partially installed".format(pkg_spec, e),
+ file=sys.stderr,
+ )
+
def get_install_path():
global install_path
@@ -242,6 +257,7 @@ def get_install_path():
install_path = expandhome(install_path)
return install_path
+
def cleanup():
for fname in cleanup_files:
try:
@@ -249,21 +265,27 @@ def cleanup():
except OSError:
print("Warning: Cannot delete " + fname)
+
def help():
- print("""\
+ print(
+ """\
upip - Simple PyPI package manager for MicroPython
Usage: micropython -m upip install [-p <path>] <package>... | -r <requirements.txt>
import upip; upip.install(package_or_list, [<path>])
If <path> is not given, packages will be installed into sys.path[1]
(can be set from MICROPYPATH environment variable, if current system
-supports that).""")
+supports that)."""
+ )
print("Current value of sys.path[1]:", sys.path[1])
- print("""\
+ print(
+ """\
Note: only MicroPython packages (usually, named micropython-*) are supported
for installation, upip does not support arbitrary code in setup.py.
-""")
+"""
+ )
+
def main():
global debug
diff --git a/tools/upip_utarfile.py b/tools/upip_utarfile.py
index 460ca2cd4..21b899f02 100644
--- a/tools/upip_utarfile.py
+++ b/tools/upip_utarfile.py
@@ -9,11 +9,12 @@ TAR_HEADER = {
DIRTYPE = "dir"
REGTYPE = "file"
+
def roundup(val, align):
return (val + align - 1) & ~(align - 1)
-class FileSection:
+class FileSection:
def __init__(self, f, content_len, aligned_len):
self.f = f
self.content_len = content_len
@@ -33,7 +34,7 @@ class FileSection:
if self.content_len == 0:
return 0
if len(buf) > self.content_len:
- buf = memoryview(buf)[:self.content_len]
+ buf = memoryview(buf)[: self.content_len]
sz = self.f.readinto(buf)
self.content_len -= sz
return sz
@@ -47,13 +48,13 @@ class FileSection:
self.f.readinto(buf, s)
sz -= s
-class TarInfo:
+class TarInfo:
def __str__(self):
return "TarInfo(%r, %s, %d)" % (self.name, self.type, self.size)
-class TarFile:
+class TarFile:
def __init__(self, name=None, fileobj=None):
if fileobj:
self.f = fileobj
@@ -62,24 +63,24 @@ class TarFile:
self.subf = None
def next(self):
- if self.subf:
- self.subf.skip()
- buf = self.f.read(512)
- if not buf:
- return None
-
- h = uctypes.struct(uctypes.addressof(buf), TAR_HEADER, uctypes.LITTLE_ENDIAN)
-
- # Empty block means end of archive
- if h.name[0] == 0:
- return None
-
- d = TarInfo()
- d.name = str(h.name, "utf-8").rstrip("\0")
- d.size = int(bytes(h.size), 8)
- d.type = [REGTYPE, DIRTYPE][d.name[-1] == "/"]
- self.subf = d.subf = FileSection(self.f, d.size, roundup(d.size, 512))
- return d
+ if self.subf:
+ self.subf.skip()
+ buf = self.f.read(512)
+ if not buf:
+ return None
+
+ h = uctypes.struct(uctypes.addressof(buf), TAR_HEADER, uctypes.LITTLE_ENDIAN)
+
+ # Empty block means end of archive
+ if h.name[0] == 0:
+ return None
+
+ d = TarInfo()
+ d.name = str(h.name, "utf-8").rstrip("\0")
+ d.size = int(bytes(h.size), 8)
+ d.type = [REGTYPE, DIRTYPE][d.name[-1] == "/"]
+ self.subf = d.subf = FileSection(self.f, d.size, roundup(d.size, 512))
+ return d
def __iter__(self):
return self