OVMS3-idf/tools/mkdfu.py

251 lines
8.1 KiB
Python
Executable file

#!/usr/bin/env python
#
# Copyright 2020 Espressif Systems (Shanghai) PTE LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# This program creates archives compatible with ESP32-S* ROM DFU implementation.
#
# The archives are in CPIO format. Each file which needs to be flashed is added to the archive
# as a separate file. In addition to that, a special index file, 'dfuinfo0.dat', is created.
# This file must be the first one in the archive. It contains binary structures describing each
# subsequent file (for example, where the file needs to be flashed/loaded).
from collections import namedtuple
from future.utils import iteritems
import argparse
import hashlib
import json
import os
import struct
import zlib
try:
import typing
except ImportError:
# Only used for type annotations
pass
try:
from itertools import izip as zip
except ImportError:
# Python 3
pass
# CPIO ("new ASCII") format related things
CPIO_MAGIC = b"070701"
CPIO_STRUCT = b"=6s" + b"8s" * 13
CPIOHeader = namedtuple(
"CPIOHeader",
[
"magic",
"ino",
"mode",
"uid",
"gid",
"nlink",
"mtime",
"filesize",
"devmajor",
"devminor",
"rdevmajor",
"rdevminor",
"namesize",
"check",
],
)
CPIO_TRAILER = "TRAILER!!!"
def make_cpio_header(
filename_len, file_len, is_trailer=False
): # type: (int, int, bool) -> CPIOHeader
""" Returns CPIOHeader for the given file name and file size """
def as_hex(val): # type: (int) -> bytes
return "{:08x}".format(val).encode("ascii")
hex_0 = as_hex(0)
mode = hex_0 if is_trailer else as_hex(0o0100644)
nlink = as_hex(1) if is_trailer else hex_0
return CPIOHeader(
magic=CPIO_MAGIC,
ino=hex_0,
mode=mode,
uid=hex_0,
gid=hex_0,
nlink=nlink,
mtime=hex_0,
filesize=as_hex(file_len),
devmajor=hex_0,
devminor=hex_0,
rdevmajor=hex_0,
rdevminor=hex_0,
namesize=as_hex(filename_len),
check=hex_0,
)
# DFU format related things
# Structure of one entry in dfuinfo0.dat
DFUINFO_STRUCT = b"<I I 64s 16s"
DFUInfo = namedtuple("DFUInfo", ["address", "flags", "name", "md5"])
DFUINFO_FILE = "dfuinfo0.dat"
# Structure which gets added at the end of the entire DFU file
DFUSUFFIX_STRUCT = b"<H H H H 3s B"
DFUSuffix = namedtuple(
"DFUSuffix", ["bcd_device", "pid", "vid", "bcd_dfu", "sig", "len"]
)
ESPRESSIF_VID = 12346
# TODO: set PID based on the chip type (add a command line argument)
DFUSUFFIX_DEFAULT = DFUSuffix(0xFFFF, 0xFFFF, ESPRESSIF_VID, 0x0100, b"UFD", 16)
# This CRC32 gets added after DFUSUFFIX_STRUCT
DFUCRC_STRUCT = b"<I"
def dfu_crc(data, crc=0): # type: (bytes, int) -> int
""" Calculate CRC32/JAMCRC of data, with an optional initial value """
uint32_max = 0xFFFFFFFF
return uint32_max - (zlib.crc32(data, crc) & uint32_max)
def pad_bytes(b, multiple, padding=b"\x00"): # type: (bytes, int, bytes) -> bytes
""" Pad 'b' to a length divisible by 'multiple' """
padded_len = (len(b) + multiple - 1) // multiple * multiple
return b + padding * (padded_len - len(b))
class EspDfuWriter(object):
def __init__(self, dest_file): # type: (typing.BinaryIO) -> None
self.dest = dest_file
self.entries = [] # type: typing.List[bytes]
self.index = [] # type: typing.List[DFUInfo]
def add_file(self, flash_addr, path): # type: (int, str) -> None
""" Add file to be written into flash at given address """
with open(path, "rb") as f:
self._add_cpio_flash_entry(os.path.basename(path), flash_addr, f.read())
def finish(self): # type: () -> None
""" Write DFU file """
# Prepare and add dfuinfo0.dat file
dfuinfo = b"".join([struct.pack(DFUINFO_STRUCT, *item) for item in self.index])
self._add_cpio_entry(DFUINFO_FILE, dfuinfo, first=True)
# Add CPIO archive trailer
self._add_cpio_entry(CPIO_TRAILER, b"", trailer=True)
# Combine all the entries and pad the file
out_data = b"".join(self.entries)
cpio_block_size = 10240
out_data = pad_bytes(out_data, cpio_block_size)
# Add DFU suffix and CRC
out_data += struct.pack(DFUSUFFIX_STRUCT, *DFUSUFFIX_DEFAULT)
out_data += struct.pack(DFUCRC_STRUCT, dfu_crc(out_data))
# Finally write the entire binary
self.dest.write(out_data)
def _add_cpio_flash_entry(
self, filename, flash_addr, data
): # type: (str, int, bytes) -> None
md5 = hashlib.md5()
md5.update(data)
self.index.append(
DFUInfo(
address=flash_addr,
flags=0,
name=filename.encode("utf-8"),
md5=md5.digest(),
)
)
self._add_cpio_entry(filename, data)
def _add_cpio_entry(
self, filename, data, first=False, trailer=False
): # type: (str, bytes, bool, bool) -> None
filename_b = filename.encode("utf-8") + b"\x00"
cpio_header = make_cpio_header(len(filename_b), len(data), is_trailer=trailer)
entry = pad_bytes(
struct.pack(CPIO_STRUCT, *cpio_header) + filename_b, 4
) + pad_bytes(data, 4)
if not first:
self.entries.append(entry)
else:
self.entries.insert(0, entry)
def action_write(args):
writer = EspDfuWriter(args['output_file'])
for addr, f in args['files']:
print('Adding {} at {:#x}'.format(f, addr))
writer.add_file(addr, f)
writer.finish()
print('"{}" has been written. You may proceed with DFU flashing.'.format(args['output_file'].name))
def main():
parser = argparse.ArgumentParser()
# Provision to add "info" command
subparsers = parser.add_subparsers(dest="command")
write_parser = subparsers.add_parser("write")
write_parser.add_argument("-o", "--output-file",
help='Filename for storing the output DFU image',
required=True,
type=argparse.FileType("wb"))
write_parser.add_argument("--json",
help='Optional file for loading "flash_files" dictionary with <address> <file> items')
write_parser.add_argument("files",
metavar="<address> <file>", help='Add <file> at <address>',
nargs="*")
args = parser.parse_args()
def check_file(file_name):
if not os.path.isfile(file_name):
raise RuntimeError('{} is not a regular file!'.format(file_name))
return file_name
files = []
if args.files:
files += [(int(addr, 0), check_file(f_name)) for addr, f_name in zip(args.files[::2], args.files[1::2])]
if args.json:
json_dir = os.path.dirname(os.path.abspath(args.json))
def process_json_file(path):
'''
The input path is relative to json_dir. This function makes it relative to the current working
directory.
'''
return check_file(os.path.relpath(os.path.join(json_dir, path), start=os.curdir))
with open(args.json) as f:
files += [(int(addr, 0),
process_json_file(f_name)) for addr, f_name in iteritems(json.load(f)['flash_files'])]
files = sorted([(addr, f_name) for addr, f_name in iteritems(dict(files))],
key=lambda x: x[0]) # remove possible duplicates and sort based on the address
cmd_args = {'output_file': args.output_file,
'files': files,
}
{'write': action_write
}[args.command](cmd_args)
if __name__ == "__main__":
main()