diff options
| author | Damien George | 2020-02-27 15:36:53 +1100 |
|---|---|---|
| committer | Damien George | 2020-02-28 10:33:03 +1100 |
| commit | 69661f3343bedf86e514337cff63d96cc42f8859 (patch) | |
| tree | af5dfb380ffdb75dda84828f63cf9d840d992f0f /tools | |
| parent | 3f39d18c2b884d32f0443e2e8114ff9d7a14d718 (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-x | tools/dfu.py | 260 | ||||
| -rw-r--r-- | tools/file2h.py | 12 | ||||
| -rw-r--r-- | tools/gen-cpydiff.py | 192 | ||||
| -rw-r--r-- | tools/gendoc.py | 245 | ||||
| -rw-r--r-- | tools/insert-usb-ids.py | 16 | ||||
| -rwxr-xr-x | tools/make-frozen.py | 11 | ||||
| -rw-r--r-- | tools/makemanifest.py | 122 | ||||
| -rwxr-xr-x | tools/metrics.py | 87 | ||||
| -rwxr-xr-x | tools/mpy-tool.py | 635 | ||||
| -rwxr-xr-x | tools/mpy_cross_all.py | 16 | ||||
| -rwxr-xr-x | tools/mpy_ld.py | 683 | ||||
| -rwxr-xr-x | tools/pyboard.py | 269 | ||||
| -rwxr-xr-x | tools/pydfu.py | 300 | ||||
| -rwxr-xr-x | tools/tinytest-codegen.py | 106 | ||||
| -rwxr-xr-x | tools/uf2conv.py | 170 | ||||
| -rw-r--r-- | tools/upip.py | 64 | ||||
| -rw-r--r-- | tools/upip_utarfile.py | 45 |
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 |
