+++ /dev/null
-#!/usr/bin/env python
-#
-# gen_otadata prints info about the otadata partition.
-#
-# Copyright 2018 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.
-from __future__ import print_function, division
-import argparse
-import os
-import re
-import struct
-import sys
-import hashlib
-import binascii
-
-__version__ = '1.0'
-
-quiet = False
-
-def status(msg):
- """ Print status message to stderr """
- if not quiet:
- critical(msg)
-
-def critical(msg):
- """ Print critical message to stderr """
- if not quiet:
- sys.stderr.write(msg)
- sys.stderr.write('\n')
-
-def little_endian(buff, offset):
- data = buff[offset:offset+4]
- data.reverse()
- data = ''.join(data)
- return data
-
-def main():
- global quiet
- parser = argparse.ArgumentParser(description='Prints otadata partition in human readable form.')
-
- parser.add_argument('--quiet', '-q', help="Don't print status messages to stderr", action='store_true')
-
- search_type = parser.add_mutually_exclusive_group()
-
- parser.add_argument('input', help='Path to binary file containing otadata partition to parse.',
- type=argparse.FileType('rb'))
-
- args = parser.parse_args()
-
- quiet = args.quiet
-
- input = args.input.read()
-
- hex_input_0 = binascii.hexlify(input)
- hex_input_0 = map(''.join, zip(*[iter(hex_input_0)]*2))
- hex_input_1 = binascii.hexlify(input[4096:])
- hex_input_1 = map(''.join, zip(*[iter(hex_input_1)]*2))
-
- print("\t%11s\t%8s |\t%8s\t%8s" %("OTA_SEQ", "CRC", "OTA_SEQ", "CRC"))
- print("Firmware: 0x%s \t 0x%s |\t0x%s \t 0x%s" % (little_endian(hex_input_0, 0), little_endian(hex_input_0, 28), \
- little_endian(hex_input_1, 0), little_endian(hex_input_1, 28)))
-class InputError(RuntimeError):
- def __init__(self, e):
- super(InputError, self).__init__(e)
-
-class ValidationError(InputError):
- def __init__(self, partition, message):
- super(ValidationError, self).__init__(
- "Partition %s invalid: %s" % (partition.name, message))
-
-if __name__ == '__main__':
- try:
- r = main()
- sys.exit(r)
- except InputError as e:
- print(e, file=sys.stderr)
- sys.exit(2)
+++ /dev/null
-#!/usr/bin/env python
-#
-# generates an empty binary file
-#
-# This tool generates an empty binary file of the required size.
-#
-# Copyright 2018 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.
-from __future__ import print_function, division
-from __future__ import unicode_literals
-import argparse
-import os
-import re
-import struct
-import sys
-import hashlib
-import binascii
-
-__version__ = '1.0'
-
-quiet = False
-
-def status(msg):
- """ Print status message to stderr """
- if not quiet:
- critical(msg)
-
-def critical(msg):
- """ Print critical message to stderr """
- if not quiet:
- sys.stderr.write(msg)
- sys.stderr.write('\n')
-
-def generate_blanked_file(size, output_path):
- output = b"\xFF" * size
- try:
- stdout_binary = sys.stdout.buffer # Python 3
- except AttributeError:
- stdout_binary = sys.stdout
- with stdout_binary if output_path == '-' else open(output_path, 'wb') as f:
- f.write(output)
-
-def main():
- global quiet
- parser = argparse.ArgumentParser(description='Generates an empty binary file of the required size.')
-
- parser.add_argument('--quiet', '-q', help="Don't print status messages to stderr", action='store_true')
-
- parser.add_argument('--size', help='Size of generated the file', type=str, required=True)
-
- parser.add_argument('output', help='Path for binary file.', nargs='?', default='-')
- args = parser.parse_args()
-
- quiet = args.quiet
-
- size = int(args.size, 0)
- if size > 0 :
- generate_blanked_file(size, args.output)
- return 0
-
-class InputError(RuntimeError):
- def __init__(self, e):
- super(InputError, self).__init__(e)
-
-if __name__ == '__main__':
- try:
- r = main()
- sys.exit(r)
- except InputError as e:
- print(e, file=sys.stderr)
- sys.exit(2)
--- /dev/null
+#!/usr/bin/env python
+#
+# otatool is used to perform ota-level operations - flashing ota partition
+# erasing ota partition and switching ota partition
+#
+# Copyright 2018 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.
+from __future__ import print_function, division
+import argparse
+import os
+import sys
+import binascii
+import subprocess
+import tempfile
+import collections
+import struct
+
+__version__ = '1.0'
+
+IDF_COMPONENTS_PATH = os.path.expandvars(os.path.join("$IDF_PATH", "components"))
+
+PARTTOOL_PY = os.path.join(IDF_COMPONENTS_PATH, "partition_table", "parttool.py")
+
+SPI_FLASH_SEC_SIZE = 0x2000
+
+quiet = False
+
+
+def status(msg):
+ if not quiet:
+ print(msg)
+
+
+def _invoke_parttool(parttool_args, args, output=False, partition=None):
+ invoke_args = []
+
+ if partition:
+ invoke_args += [sys.executable, PARTTOOL_PY] + partition
+ else:
+ invoke_args += [sys.executable, PARTTOOL_PY, "--partition-type", "data", "--partition-subtype", "ota"]
+
+ if quiet:
+ invoke_args += ["-q"]
+
+ if args.port != "":
+ invoke_args += ["--port", args.port]
+
+ if args.partition_table_file:
+ invoke_args += ["--partition-table-file", args.partition_table_file]
+
+ if args.partition_table_offset:
+ invoke_args += ["--partition-table-offset", args.partition_table_offset]
+
+ invoke_args += parttool_args
+
+ if output:
+ return subprocess.check_output(invoke_args)
+ else:
+ return subprocess.check_call(invoke_args)
+
+
+def _get_otadata_contents(args, check=True):
+ global quiet
+
+ if check:
+ check_args = ["get_partition_info", "--info", "offset", "size"]
+
+ quiet = True
+ output = _invoke_parttool(check_args, args, True).split(b" ")
+ quiet = args.quiet
+
+ if not output:
+ raise RuntimeError("No ota_data partition found")
+
+ with tempfile.NamedTemporaryFile() as otadata_file:
+ invoke_args = ["read_partition", "--output", otadata_file.name]
+ _invoke_parttool(invoke_args, args)
+ return otadata_file.read()
+
+
+def _get_otadata_status(otadata_contents):
+ status = []
+
+ otadata_status = collections.namedtuple("otadata_status", "seq crc")
+
+ for i in range(2):
+ start = i * (SPI_FLASH_SEC_SIZE >> 1)
+
+ seq = bytearray(otadata_contents[start:start + 4])
+ crc = bytearray(otadata_contents[start + 28:start + 32])
+
+ seq = struct.unpack('>I', seq)
+ crc = struct.unpack('>I', crc)
+
+ status.append(otadata_status(seq[0], crc[0]))
+
+ return status
+
+
+def read_otadata(args):
+ status("Reading ota_data partition contents...")
+ otadata_info = _get_otadata_contents(args)
+ otadata_info = _get_otadata_status(otadata_info)
+
+ print(otadata_info)
+
+ print("\t\t{:11}\t{:8s}|\t{:8s}\t{:8s}".format("OTA_SEQ", "CRC", "OTA_SEQ", "CRC"))
+ print("Firmware: 0x{:8x} \t 0x{:8x} |\t0x{:8x} \t 0x{:8x}".format(otadata_info[0].seq, otadata_info[0].crc,
+ otadata_info[1].seq, otadata_info[1].crc))
+
+
+def erase_otadata(args):
+ status("Erasing ota_data partition contents...")
+ _invoke_parttool(["erase_partition"], args)
+ status("Erased ota_data partition contents")
+
+
+def switch_otadata(args):
+ sys.path.append(os.path.join(IDF_COMPONENTS_PATH, "partition_table"))
+ import gen_esp32part as gen
+
+ def is_otadata_status_valid(status):
+ seq = status.seq % (1 << 32)
+ crc = hex(binascii.crc32(struct.pack("I", seq), 0xFFFFFFFF) % (1 << 32))
+ return seq < (int('0xFFFFFFFF', 16) % (1 << 32)) and status.crc == crc
+
+ status("Looking for ota app partitions...")
+
+ # In order to get the number of ota app partitions, we need the partition table
+ partition_table = None
+
+ with tempfile.NamedTemporaryFile() as partition_table_file:
+ invoke_args = ["get_partition_info", "--table", partition_table_file.name]
+
+ _invoke_parttool(invoke_args, args)
+
+ partition_table = partition_table_file.read()
+ partition_table = gen.PartitionTable.from_binary(partition_table)
+
+ ota_partitions = list()
+
+ for i in range(gen.NUM_PARTITION_SUBTYPE_APP_OTA):
+ ota_partition = filter(lambda p: p.subtype == (gen.MIN_PARTITION_SUBTYPE_APP_OTA + i), partition_table)
+
+ try:
+ ota_partitions.append(list(ota_partition)[0])
+ except IndexError:
+ break
+
+ ota_partitions = sorted(ota_partitions, key=lambda p: p.subtype)
+
+ if not ota_partitions:
+ raise RuntimeError("No ota app partitions found")
+
+ status("Verifying partition to switch to exists...")
+
+ # Look for the app partition to switch to
+ ota_partition_next = None
+
+ try:
+ if args.name:
+ ota_partition_next = filter(lambda p: p.name == args.name, ota_partitions)
+ else:
+ ota_partition_next = filter(lambda p: p.subtype - gen.MIN_PARTITION_SUBTYPE_APP_OTA == args.slot, ota_partitions)
+
+ ota_partition_next = list(ota_partition_next)[0]
+ except IndexError:
+ raise RuntimeError("Partition to switch to not found")
+
+ otadata_contents = _get_otadata_contents(args)
+ otadata_status = _get_otadata_status(otadata_contents)
+
+ # Find the copy to base the computation for ota sequence number on
+ otadata_compute_base = -1
+
+ # Both are valid, take the max as computation base
+ if is_otadata_status_valid(otadata_status[0]) and is_otadata_status_valid(otadata_status[1]):
+ if otadata_status[0].seq >= otadata_status[1].seq:
+ otadata_compute_base = 0
+ else:
+ otadata_compute_base = 1
+ # Only one copy is valid, use that
+ elif is_otadata_status_valid(otadata_status[0]):
+ otadata_compute_base = 0
+ elif is_otadata_status_valid(otadata_status[1]):
+ otadata_compute_base = 1
+ # Both are invalid (could be initial state - all 0xFF's)
+ else:
+ pass
+
+ ota_seq_next = 0
+ ota_partitions_num = len(ota_partitions)
+
+ target_seq = (ota_partition_next.subtype & 0x0F) + 1
+
+ # Find the next ota sequence number
+ if otadata_compute_base == 0 or otadata_compute_base == 1:
+ base_seq = otadata_status[otadata_compute_base].seq % (1 << 32)
+
+ i = 0
+ while base_seq > target_seq % ota_partitions_num + i * ota_partitions_num:
+ i += 1
+
+ ota_seq_next = target_seq % ota_partitions_num + i * ota_partitions_num
+ else:
+ ota_seq_next = target_seq
+
+ # Create binary data from computed values
+ ota_seq_next = struct.pack("I", ota_seq_next)
+ ota_seq_crc_next = binascii.crc32(ota_seq_next, 0xFFFFFFFF) % (1 << 32)
+ ota_seq_crc_next = struct.pack("I", ota_seq_crc_next)
+
+ with tempfile.NamedTemporaryFile() as otadata_next_file:
+ start = (1 if otadata_compute_base == 0 else 0) * (SPI_FLASH_SEC_SIZE >> 1)
+
+ otadata_next_file.write(otadata_contents)
+
+ otadata_next_file.seek(start)
+ otadata_next_file.write(ota_seq_next)
+
+ otadata_next_file.seek(start + 28)
+ otadata_next_file.write(ota_seq_crc_next)
+
+ otadata_next_file.flush()
+
+ _invoke_parttool(["write_partition", "--input", otadata_next_file.name], args)
+ status("Updated ota_data partition")
+
+
+def _get_partition_specifier(args):
+ if args.name:
+ return ["--partition-name", args.name]
+ else:
+ return ["--partition-type", "app", "--partition-subtype", "ota_" + str(args.slot)]
+
+
+def read_ota_partition(args):
+ invoke_args = ["read_partition", "--output", args.output]
+ _invoke_parttool(invoke_args, args, partition=_get_partition_specifier(args))
+ status("Read ota partition contents to file {}".format(args.output))
+
+
+def write_ota_partition(args):
+ invoke_args = ["write_partition", "--input", args.input]
+ _invoke_parttool(invoke_args, args, partition=_get_partition_specifier(args))
+ status("Written contents of file {} to ota partition".format(args.input))
+
+
+def erase_ota_partition(args):
+ invoke_args = ["erase_partition"]
+ _invoke_parttool(invoke_args, args, partition=_get_partition_specifier(args))
+ status("Erased contents of ota partition")
+
+
+def main():
+ global quiet
+
+ parser = argparse.ArgumentParser("ESP-IDF OTA Partitions Tool")
+
+ parser.add_argument("--quiet", "-q", help="suppress stderr messages", action="store_true")
+
+ # There are two possible sources for the partition table: a device attached to the host
+ # or a partition table CSV/binary file. These sources are mutually exclusive.
+ partition_table_info_source_args = parser.add_mutually_exclusive_group()
+
+ partition_table_info_source_args.add_argument("--port", "-p", help="port where the device to read the partition table from is attached", default="")
+ partition_table_info_source_args.add_argument("--partition-table-file", "-f", help="file (CSV/binary) to read the partition table from", default="")
+
+ parser.add_argument("--partition-table-offset", "-o", help="offset to read the partition table from", default="0x8000")
+
+ subparsers = parser.add_subparsers(dest="operation", help="run otatool -h for additional help")
+
+ # Specify the supported operations
+ subparsers.add_parser("read_otadata", help="read otadata partition")
+ subparsers.add_parser("erase_otadata", help="erase otadata partition")
+
+ slot_or_name_parser = argparse.ArgumentParser(add_help=False)
+ slot_or_name_parser_args = slot_or_name_parser.add_mutually_exclusive_group()
+ slot_or_name_parser_args.add_argument("--slot", help="slot number of the ota partition", type=int)
+ slot_or_name_parser_args.add_argument("--name", help="name of the ota partition")
+
+ subparsers.add_parser("switch_otadata", help="switch otadata partition", parents=[slot_or_name_parser])
+
+ read_ota_partition_subparser = subparsers.add_parser("read_ota_partition", help="read contents of an ota partition", parents=[slot_or_name_parser])
+ read_ota_partition_subparser.add_argument("--output", help="file to write the contents of the ota partition to")
+
+ write_ota_partition_subparser = subparsers.add_parser("write_ota_partition", help="write contents to an ota partition", parents=[slot_or_name_parser])
+ write_ota_partition_subparser.add_argument("--input", help="file whose contents to write to the ota partition")
+
+ subparsers.add_parser("erase_ota_partition", help="erase contents of an ota partition", parents=[slot_or_name_parser])
+
+ args = parser.parse_args()
+
+ quiet = args.quiet
+
+ # No operation specified, display help and exit
+ if args.operation is None:
+ if not quiet:
+ parser.print_help()
+ sys.exit(1)
+
+ # Else execute the operation
+ operation_func = globals()[args.operation]
+
+ if quiet:
+ # If exceptions occur, suppress and exit quietly
+ try:
+ operation_func(args)
+ except Exception:
+ sys.exit(2)
+ else:
+ operation_func(args)
+
+
+if __name__ == '__main__':
+ main()
projects / esp-idf / commitdiff