From ebc2d658d8ebc352f013505f9eb47fce15eb424b Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Tue, 30 Jun 2026 11:57:04 +1000 Subject: [PATCH 1/4] tools: cloud: kv_state: print raw hex array Add an option to `infuse cloud device kv_state` to display the raw hex value of the key, instead of the decoded data. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/cloud.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index b92e930..0e070b8 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -5,6 +5,7 @@ __author__ = "Jordan Yates" __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd" +import base64 import glob import sys from typing import Any @@ -193,6 +194,7 @@ def add_parser(cls, parser): kv_parser.set_defaults(command_fn=cls.kv_state) kv_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID") kv_parser.add_argument("--schedules", action="store_true", help="Display task schedules") + kv_parser.add_argument("--hex", action="store_true", help="Display values as hex strings instead of decoding") dfu_parser = tool_parser.add_parser("dfu", help="Manage device firmware upgrades") dfu_parser.set_defaults(command_fn=cls.dfu) @@ -315,10 +317,13 @@ def kv_state(self, client: Client): else: table.append((key, "Not set")) else: - if isinstance(element.decoded, Unset): - table.append((key, element.data)) + if self.args.hex: + table.append((key, base64.b64decode(element.data).hex())) else: - self._kv_display(table, key, element.decoded.additional_properties) + if isinstance(element.decoded, Unset): + table.append((key, element.data)) + else: + self._kv_display(table, key, element.decoded.additional_properties) print(tabulate(table)) From 97f9e4097b31a471a21c82ed2849fa56a6697af0 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Tue, 30 Jun 2026 12:05:37 +1000 Subject: [PATCH 2/4] tools: cloud: kv_state: print key IDs Display the key IDs in the `infuse cloud device kv_state` output. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/cloud.py | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index 0e070b8..2abd2aa 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -287,12 +287,13 @@ def val_or_na(value) -> str: print(tabulate(table)) - def _kv_display(self, table: list[tuple[str, Any]], name_base: str, dictionary: dict): + def _kv_display(self, table: list[tuple[str, str, Any]], key_val: str, name_base: str, dictionary: dict): for name, value in dictionary.items(): if isinstance(value, dict): - self._kv_display(table, f"{name_base}.{name}", value) + self._kv_display(table, key_val, f"{name_base}.{name}", value) else: - table.append((f"{name_base}.{name}", value)) + table.append((key_val, f"{name_base}.{name}", value)) + key_val = "" def kv_state(self, client: Client): id_int = int(self.args.id, 0) @@ -303,9 +304,10 @@ def kv_state(self, client: Client): print(f"Unable to query KV state for {id_str}") return - table: list[tuple[str, Any]] = [] + table: list[tuple[str, str, Any]] = [] for element in kv_state: - key = element.key_name if isinstance(element.key_name, str) else str(element.key_id) + key = element.key_name if isinstance(element.key_name, str) else "" + key_id = str(element.key_id) # Don't display task schedules unless requested if key == "TASK_SCHEDULES" and not self.args.schedules: @@ -313,17 +315,17 @@ def kv_state(self, client: Client): if isinstance(element.data, Unset): if element.crc: - table.append((key, f"Write-only (CRC: 0x{element.crc:08x})")) + table.append((key_id, key, f"Write-only (CRC: 0x{element.crc:08x})")) else: - table.append((key, "Not set")) + table.append((key_id, key, "Not set")) else: if self.args.hex: - table.append((key, base64.b64decode(element.data).hex())) + table.append((key_id, key, base64.b64decode(element.data).hex())) else: if isinstance(element.decoded, Unset): - table.append((key, element.data)) + table.append((key_id, key, element.data)) else: - self._kv_display(table, key, element.decoded.additional_properties) + self._kv_display(table, key_id, key, element.decoded.additional_properties) print(tabulate(table)) From 377c2aa4fec9362d7e8d0155cc5d6ef1b52c2fe7 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Tue, 30 Jun 2026 13:36:15 +1000 Subject: [PATCH 3/4] tools: cloud: kv_state: print base64 output Add an option to `infuse cloud device kv_state` to display the base64 encoded value of the key, instead of the decoded data. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/cloud.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/infuse_iot/tools/cloud.py b/src/infuse_iot/tools/cloud.py index 2abd2aa..dd5ea42 100644 --- a/src/infuse_iot/tools/cloud.py +++ b/src/infuse_iot/tools/cloud.py @@ -194,7 +194,11 @@ def add_parser(cls, parser): kv_parser.set_defaults(command_fn=cls.kv_state) kv_parser.add_argument("--id", type=str, required=True, help="Infuse-IoT device ID") kv_parser.add_argument("--schedules", action="store_true", help="Display task schedules") - kv_parser.add_argument("--hex", action="store_true", help="Display values as hex strings instead of decoding") + kv_display = kv_parser.add_mutually_exclusive_group() + kv_display.add_argument("--hex", action="store_true", help="Display values as hex strings instead of decoding") + kv_display.add_argument( + "--base64", action="store_true", help="Display values as base64 strings instead of decoding" + ) dfu_parser = tool_parser.add_parser("dfu", help="Manage device firmware upgrades") dfu_parser.set_defaults(command_fn=cls.dfu) @@ -321,6 +325,8 @@ def kv_state(self, client: Client): else: if self.args.hex: table.append((key_id, key, base64.b64decode(element.data).hex())) + elif self.args.base64: + table.append((key_id, key, element.data)) else: if isinstance(element.decoded, Unset): table.append((key_id, key, element.data)) From feda087d460b8ce378cfea9d3dde3b7e4e7ba98b Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Tue, 30 Jun 2026 13:38:01 +1000 Subject: [PATCH 4/4] scripts: elf_task_schedules: added Add a script which finds the default schedules in a Zephyr `.elf` file and outputs them to the terminal. This can simplify the process of generating the base64 encoded string needed to update the KV store on the cloud. 1. Recompile application with desired new schedule 2. Run `elf_task_schedules.py --base64` to get the value to set Signed-off-by: Jordan Yates --- scripts/elf_task_schedules.py | 333 ++++++++++++++++++++++++++++++++++ 1 file changed, 333 insertions(+) create mode 100755 scripts/elf_task_schedules.py diff --git a/scripts/elf_task_schedules.py b/scripts/elf_task_schedules.py new file mode 100755 index 0000000..9639053 --- /dev/null +++ b/scripts/elf_task_schedules.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 + +import argparse +import base64 +import sys +from dataclasses import dataclass +from pathlib import Path + +from elftools.dwarf.descriptions import describe_form_class +from elftools.dwarf.dwarf_expr import DWARFExprParser +from elftools.dwarf.locationlists import LocationEntry +from elftools.elf.elffile import ELFFile + +STRUCT_NAME = "task_schedule" + + +@dataclass +class Candidate: + name: str + address: int + size: int + element_size: int + file_offset: int + source: str | None + + +@dataclass +class SkippedCandidate: + name: str + reason: str + source: str | None + + +def die_name(die): + attr = die.attributes.get("DW_AT_name") + if attr is None: + return None + return attr.value.decode("utf-8", errors="replace") + + +def type_die(die): + if "DW_AT_type" not in die.attributes: + return None + return die.get_DIE_from_attribute("DW_AT_type") + + +def resolve_type(die): + while die is not None and die.tag in ( + "DW_TAG_const_type", + "DW_TAG_volatile_type", + "DW_TAG_restrict_type", + "DW_TAG_atomic_type", + "DW_TAG_typedef", + ): + die = type_die(die) + return die + + +def type_size(die): + die = resolve_type(die) + if die is None: + return None + if "DW_AT_byte_size" in die.attributes: + return die.attributes["DW_AT_byte_size"].value + if die.tag == "DW_TAG_array_type": + return array_size(die) + return None + + +def subrange_count(subrange_die): + if "DW_AT_count" in subrange_die.attributes: + return subrange_die.attributes["DW_AT_count"].value + if "DW_AT_upper_bound" not in subrange_die.attributes: + return None + + upper = subrange_die.attributes["DW_AT_upper_bound"].value + lower = subrange_die.attributes.get("DW_AT_lower_bound") + lower = lower.value if lower is not None else 0 + return upper - lower + 1 + + +def array_size(array_die): + if "DW_AT_byte_size" in array_die.attributes: + return array_die.attributes["DW_AT_byte_size"].value + + elem = type_size(type_die(array_die)) + if elem is None: + return None + + count = 1 + for child in array_die.iter_children(): + if child.tag != "DW_TAG_subrange_type": + continue + sub_count = subrange_count(child) + if sub_count is None: + return None + count *= sub_count + return elem * count + + +def array_element_size(array_die): + return type_size(type_die(array_die)) + + +def is_task_schedule_array(var_die): + var_type = resolve_type(type_die(var_die)) + if var_type is None or var_type.tag != "DW_TAG_array_type": + return False + + elem_type = resolve_type(type_die(var_type)) + return elem_type is not None and elem_type.tag == "DW_TAG_structure_type" and die_name(elem_type) == STRUCT_NAME + + +def source_location(die, dwarf_info): + file_attr = die.attributes.get("DW_AT_decl_file") + if file_attr is None: + return None + + lineprog = dwarf_info.line_program_for_CU(die.cu) + if lineprog is None: + return None + + file_index = file_attr.value - 1 + if file_index < 0 or file_index >= len(lineprog["file_entry"]): + return None + + entry = lineprog["file_entry"][file_index] + filename = entry.name.decode("utf-8", errors="replace") + include_dirs = lineprog.header.include_directory + directory = "" + if entry.dir_index != 0: + directory = include_dirs[entry.dir_index - 1].decode("utf-8", errors="replace") + + line = die.attributes.get("DW_AT_decl_line") + path = str(Path(directory) / filename) if directory else filename + return f"{path}:{line.value}" if line is not None else path + + +def address_from_expr(expr, dwarf_info): + parser = DWARFExprParser(dwarf_info.structs) + operations = parser.parse_expr(expr) + if len(operations) != 1: + return None + + op = operations[0] + if op.op_name == "DW_OP_addr": + return op.args[0] + + # Some toolchains emit indexed address operations in split-DWARF-like forms. + if op.op_name in ("DW_OP_addrx", "DW_OP_GNU_addr_index"): + try: + return dwarf_info.get_addr(op.args[0]) + except AttributeError: + return None + + return None + + +def variable_address(var_die, dwarf_info): + loc_attr = var_die.attributes.get("DW_AT_location") + if loc_attr is None: + return None + + form_class = describe_form_class(loc_attr.form) + if form_class == "exprloc": + return address_from_expr(loc_attr.value, dwarf_info) + + if form_class != "loclist": + return None + + loc_lists = dwarf_info.location_lists() + loc_list = loc_lists.get_location_list_at_offset(loc_attr.value, die=var_die) + addresses = set() + for entry in loc_list: + if isinstance(entry, LocationEntry): + address = address_from_expr(entry.loc_expr, dwarf_info) + if address is not None: + addresses.add(address) + + return addresses.pop() if len(addresses) == 1 else None + + +def file_offset_for_address(elf, address, size): + for segment in elf.iter_segments(): + if segment.header.p_type != "PT_LOAD": + continue + + start = segment.header.p_vaddr + file_end = start + segment.header.p_filesz + if start <= address and address + size <= file_end: + return segment.header.p_offset + (address - start) + + for section in elf.iter_sections(): + if section.header.sh_type == "SHT_NOBITS" or section.header.sh_addr == 0: + continue + + start = section.header.sh_addr + end = start + section.header.sh_size + if start <= address and address + size <= end: + return section.header.sh_offset + (address - start) + + return None + + +def symbol_file_offsets(elf, name, size): + offsets = [] + for section in elf.iter_sections(): + if section.header.sh_type not in ("SHT_SYMTAB", "SHT_DYNSYM"): + continue + + for symbol in section.iter_symbols(): + if symbol.name != name: + continue + if symbol["st_info"]["type"] != "STT_OBJECT": + continue + if isinstance(symbol["st_shndx"], str): + continue + + target = elf.get_section(symbol["st_shndx"]) + if target.header.sh_type == "SHT_NOBITS": + continue + + sym_size = symbol["st_size"] + if sym_size != 0 and sym_size < size: + continue + + if target.header.sh_addr == 0: + offset = target.header.sh_offset + symbol["st_value"] + else: + offset = target.header.sh_offset + symbol["st_value"] - target.header.sh_addr + + section_start = target.header.sh_offset + section_end = section_start + target.header.sh_size + if section_start <= offset and offset + size <= section_end: + offsets.append(offset) + + return sorted(set(offsets)) + + +def find_candidates(elf, dwarf_info, name_filter): + candidates = [] + skipped = [] + + for cu in dwarf_info.iter_CUs(): + for die in cu.iter_DIEs(): + if die.tag != "DW_TAG_variable" or not is_task_schedule_array(die): + continue + + name = die_name(die) or "" + source = source_location(die, dwarf_info) + if name_filter is not None and name != name_filter: + continue + + array_die = resolve_type(type_die(die)) + size = array_size(array_die) + if size is None: + skipped.append(SkippedCandidate(name, "unknown array size", source)) + continue + + element_size = array_element_size(array_die) + if element_size is None: + skipped.append(SkippedCandidate(name, "unknown array element size", source)) + continue + + address = variable_address(die, dwarf_info) + if address is None: + skipped.append(SkippedCandidate(name, "no absolute ELF address", source)) + continue + + file_offset = file_offset_for_address(elf, address, size) + if file_offset is None: + offsets = symbol_file_offsets(elf, name, size) + if len(offsets) == 1: + file_offset = offsets[0] + else: + skipped.append(SkippedCandidate(name, "not backed by initialized ELF bytes", source)) + continue + + candidates.append(Candidate(name, address, size, element_size, file_offset, source)) + + return candidates, skipped + + +def describe_candidate(candidate): + source = f" ({candidate.source})" if candidate.source else "" + return f"{candidate.name} @ 0x{candidate.address:x}, {candidate.size} bytes{source}" + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Print the raw hex bytes for the single initialized 'struct task_schedule' array in a Zephyr ELF.", + allow_abbrev=False, + ) + parser.add_argument("elf", type=Path, help="Path to zephyr.elf") + parser.add_argument( + "--name", + help="Require a specific variable name when the ELF contains multiple initialized arrays", + ) + parser.add_argument("--base64", action="store_true", help="Output base64 data instead of hex") + args = parser.parse_args() + + with args.elf.open("rb") as f: + elf = ELFFile(f) + if not elf.has_dwarf_info(): + sys.exit(f"{args.elf}: no DWARF debug info found") + + dwarf_info = elf.get_dwarf_info() + candidates, skipped = find_candidates(elf, dwarf_info, args.name) + + if len(candidates) != 1: + details = [] + if candidates: + details.append("initialized candidates:") + details.extend(f" - {describe_candidate(c)}" for c in candidates) + if skipped: + details.append("skipped candidates:") + details.extend(f" - {s.name}: {s.reason}" + (f" ({s.source})" if s.source else "") for s in skipped) + hint = " Use --name to select one." if len(candidates) > 1 else "" + sys.exit( + f"expected exactly one initialized struct {STRUCT_NAME} array, " + f"found {len(candidates)}.{hint}\n" + "\n".join(details) + ) + + candidate = candidates[0] + f.seek(candidate.file_offset) + data = f.read(candidate.size) + + array_elements = [data[i : i + candidate.element_size] for i in range(0, len(data), candidate.element_size)] + + def fmt(val: bytes): + return base64.b64encode(val).decode("utf-8") if args.base64 else val.hex() + + print("\n".join(f"{1001 + i:3d}: {fmt(val)}" for i, val in enumerate(array_elements)))