From 0d77a682be3cda2bd011ac91788d6138f459740d Mon Sep 17 00:00:00 2001 From: Aeyohan Furtado Date: Mon, 13 Jul 2026 11:42:13 +1000 Subject: [PATCH 1/6] tools: ota_upgrade: cross-app upgrade support Tool had partial support to update devices using diffs of a different application id. Formally extended support for cross-application upgrades by: * Disabling using cross-app diff by default, and enabling with cli arg. * When enabled, allowing devices with a different app-id to be upgraded using a diff from a different app-id Usage: add `--cross-app` to `ota_upgrade.py --cross-app -r ...` to allow devices running a different app to be upgraded. Notes: * `--cross-app` is not required when the diff selected by `--single` is already for a different application id. * As a result, it has no effect when used with `--single`. * Requires a diff to be already generated for the release. This does not create one. --- src/infuse_iot/tools/ota_upgrade.py | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index ebed674..47ef9e2 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -44,9 +44,21 @@ def __init__(self, args): self._conn_timeout = args.conn_timeout self._min_rssi: int | None = args.rssi self._explicit_ids: list[int] = [] + self._supported_apps: list[int] = [] if args.release: self._release: ValidRelease = args.release self._single_diff = None + # Also capture any releases for other applications. + if args.cross_app: + release_dir = self._release.dir / "diffs" + # List out *.bin files in release_dir and filter where the parent folder is not the release_dir itself + for diff_file in release_dir.glob("**/*.bin"): + if diff_file.parent != release_dir and diff_file.parent.parent == release_dir: + # This is a diff for a different application + diff_app_id = int(diff_file.parent.name, 0) + if diff_app_id not in self._supported_apps: + self._supported_apps.append(diff_app_id) + elif args.single: # Find the associated release diff_folder = args.single.parent @@ -58,6 +70,12 @@ def __init__(self, args): release_folder = diff_folder.parent self._release = ValidRelease(str(release_folder)) self._single_diff = args.single + + diff_app_id = int(args.single.parent.name, 0) + if diff_app_id != self._release.metadata["application"]: + self._supported_apps.append(diff_app_id) + if args.cross_app and diff_app_id not in self._supported_apps: + self._supported_apps.append(diff_app_id) else: raise NotImplementedError("Unknow upgrade type") app_meta = self._release.metadata["application"] @@ -97,6 +115,7 @@ def add_parser(cls, parser): upgrade_type = parser.add_mutually_exclusive_group(required=True) upgrade_type.add_argument("--release", "-r", type=ValidRelease, help="Application release to upgrade to") upgrade_type.add_argument("--single", type=ValidFile, help="Single diff") + parser.add_argument("--cross-app", action="store_true", help="Allow upgrades from other applications") parser.add_argument("--rssi", type=int, help="Minimum RSSI to attempt upgrade process") parser.add_argument("--log", type=str, help="File to write upgrade results to") parser.add_argument( @@ -226,7 +245,7 @@ def run(self): self.state_update(live, "All devices updated") return else: - if announce.application != self._app_id: + if announce.application != self._app_id and announce.application not in self._supported_apps: continue if source.infuse_id in self._handled: continue @@ -256,7 +275,7 @@ def run(self): continue # Already running the requested version? - if v_str == self._new_ver: + if v_str == self._new_ver and announce.application == self._app_id: self._handled.append(source.infuse_id) self._already += 1 self.state_update(live, "Scanning") @@ -270,7 +289,7 @@ def run(self): # Do we have a valid diff? diff_file = self._release.dir / "diffs" / f"{v_str}.bin" - if not diff_file.exists(): + if not diff_file.exists() and announce.application in self._supported_apps: # Is this a single diff from a different application we know about? diff_file = self._release.dir / "diffs" / f"0x{announce.application:08x}" / f"{v_str}.bin" if not diff_file.exists(): From 8375e695d1740beb4b99580c80fa64a767088f45 Mon Sep 17 00:00:00 2001 From: Aeyohan Furtado Date: Mon, 13 Jul 2026 18:10:06 +1000 Subject: [PATCH 2/6] tools: ota_upgrade: upload failure error handling Improved patch upload failure error handing. When uploading a patch fails, prints out an error message. e.g. when using `--single`, if the gateway and upgrade target are on different networks, The RPC fails with `-EIO`. The generic error is now printed to the console. --- src/infuse_iot/tools/ota_upgrade.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index 47ef9e2..545a5fb 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -226,6 +226,9 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived): self._failed += 1 elif hdr.return_code == 0: self._pending[source.infuse_id] = time.time() + 60 + elif hdr.return_code < 0: + err = errno.strerror(-hdr.return_code) + print(f"Failed to copy patch file to {source.infuse_id:016X} ({err})") def run(self): if not self._client.comms_check(): From ab10babafe1b3f8995da7624dbfec47901162cac Mon Sep 17 00:00:00 2001 From: Aeyohan Furtado Date: Tue, 14 Jul 2026 10:58:04 +1000 Subject: [PATCH 3/6] tools: ota_upgrade: decouple run loop Decoupled update logic from main run loop. This will allow the update loop to be run on multiple threads. Signed-off-by: Aeyohan Furtado --- src/infuse_iot/tools/ota_upgrade.py | 197 ++++++++++++++-------------- 1 file changed, 100 insertions(+), 97 deletions(-) diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index 545a5fb..ae01ce4 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -230,112 +230,115 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived): err = errno.strerror(-hdr.return_code) print(f"Failed to copy patch file to {source.infuse_id:016X} ({err})") - def run(self): - if not self._client.comms_check(): - sys.exit("No communications gateway detected (infuse gateway/bt_native)") - - if self._single_diff: - self.gateway_diff_load() - - with Live(self.progress_table(), refresh_per_second=4) as live: - for source, announce in self._client.observe_announce(): - self.state_update(live, "Scanning") - if len(self._explicit_ids): - if source.infuse_id not in self._explicit_ids: - continue - if len(self._handled) == len(self._explicit_ids): - # We've handled all devices - self.state_update(live, "All devices updated") - return - else: - if announce.application != self._app_id and announce.application not in self._supported_apps: - continue - if source.infuse_id in self._handled: + def run_thread(self, live: Live): + for source, announce in self._client.observe_announce(): + self.state_update(live, "Scanning") + if len(self._explicit_ids): + if source.infuse_id not in self._explicit_ids: continue - if isinstance(announce, readings.announce_v2) and announce.board_crc != self._board_crc: + if len(self._handled) == len(self._explicit_ids): + # We've handled all devices + self.state_update(live, "All devices updated") + return + else: + if announce.application != self._app_id and announce.application not in self._supported_apps: continue - v = announce.version - v_str = f"{v.major}.{v.minor}.{v.revision}+{v.build_num:08x}" - - # Check against pending upgrades - if source.infuse_id in self._pending: - if (v_str != self._new_ver) and (time.time() < self._pending[source.infuse_id]): - # Device could still be applying the upgrade - continue - self._pending.pop(source.infuse_id) - self._handled.append(source.infuse_id) - if v_str == self._new_ver: - self._updated += 1 - result = "upgraded" - else: - self._failed += 1 - result = "failed" - if self._log: - self._log.write( - f"{time.time()},0x{source.infuse_id:016x},0x{self._app_id:08x},{v_str},{result}\n" - ) - self._log.flush() + if source.infuse_id in self._handled: + continue + if isinstance(announce, readings.announce_v2) and announce.board_crc != self._board_crc: + continue + v = announce.version + v_str = f"{v.major}.{v.minor}.{v.revision}+{v.build_num:08x}" + + # Check against pending upgrades + if source.infuse_id in self._pending: + if (v_str != self._new_ver) and (time.time() < self._pending[source.infuse_id]): + # Device could still be applying the upgrade continue - - # Already running the requested version? - if v_str == self._new_ver and announce.application == self._app_id: - self._handled.append(source.infuse_id) - self._already += 1 - self.state_update(live, "Scanning") - if self._log: - self._log.write( - f"{time.time()},0x{source.infuse_id:016x},0x{self._app_id:08x},{v_str},already\n" - ) - self._log.flush() - continue - - # Do we have a valid diff? - diff_file = self._release.dir / "diffs" / f"{v_str}.bin" - - if not diff_file.exists() and announce.application in self._supported_apps: - # Is this a single diff from a different application we know about? - diff_file = self._release.dir / "diffs" / f"0x{announce.application:08x}" / f"{v_str}.bin" - if not diff_file.exists(): - self._missing_diffs.add(v_str) - self._handled.append(source.infuse_id) - self._no_diff += 1 - self.state_update(live, "Scanning") - continue - - if self._single_diff and self._single_diff != diff_file: - # Not the file we've copied to the gateway flash + self._pending.pop(source.infuse_id) + self._handled.append(source.infuse_id) + if v_str == self._new_ver: + self._updated += 1 + result = "upgraded" + else: + self._failed += 1 + result = "failed" + if self._log: + self._log.write( + f"{time.time()},0x{source.infuse_id:016x},0x{self._app_id:08x},{v_str},{result}\n" + ) + self._log.flush() + continue + + # Already running the requested version? + if v_str == self._new_ver and announce.application == self._app_id: + self._handled.append(source.infuse_id) + self._already += 1 + self.state_update(live, "Scanning") + if self._log: + self._log.write( + f"{time.time()},0x{source.infuse_id:016x},0x{self._app_id:08x},{v_str},already\n" + ) + self._log.flush() + continue + + # Do we have a valid diff? + diff_file = self._release.dir / "diffs" / f"{v_str}.bin" + + if not diff_file.exists() and announce.application in self._supported_apps: + # Is this a single diff from a different application we know about? + diff_file = self._release.dir / "diffs" / f"0x{announce.application:08x}" / f"{v_str}.bin" + if not diff_file.exists(): self._missing_diffs.add(v_str) self._handled.append(source.infuse_id) self._no_diff += 1 self.state_update(live, "Scanning") continue - # Is signal strong enough to connect? - if self._min_rssi and source.rssi < self._min_rssi: - continue + if self._single_diff and self._single_diff != diff_file: + # Not the file we've copied to the gateway flash + self._missing_diffs.add(v_str) + self._handled.append(source.infuse_id) + self._no_diff += 1 + self.state_update(live, "Scanning") + continue + + # Is signal strong enough to connect? + if self._min_rssi and source.rssi < self._min_rssi: + continue + + # Load patch file + with open(diff_file, "rb") as f: + self.patch_file = f.read() + + # Attempt to upload + self.state_update(live, f"Connecting to {source.infuse_id:016X}") + try: + with self._client.connection( + source.infuse_id, GatewayRequestConnectionRequest.DataType.COMMAND, self._conn_timeout + ) as mtu: + if self._single_diff: + self.run_file_copy(live, mtu, source) + else: + self.run_file_upload(live, mtu, source) - # Load patch file - with open(diff_file, "rb") as f: - self.patch_file = f.read() - - # Attempt to upload - self.state_update(live, f"Connecting to {source.infuse_id:016X}") - try: - with self._client.connection( - source.infuse_id, GatewayRequestConnectionRequest.DataType.COMMAND, self._conn_timeout - ) as mtu: - if self._single_diff: - self.run_file_copy(live, mtu, source) - else: - self.run_file_upload(live, mtu, source) - - except ConnectionRefusedError: - self.state_update(live, "Scanning") - except ConnectionAbortedError: - self.state_update(live, "Scanning") + except ConnectionRefusedError: + self.state_update(live, "Scanning") + except ConnectionAbortedError: + self.state_update(live, "Scanning") - if self.task is not None: - self.progress.remove_task(self.task) - self.task = None + if self.task is not None: + self.progress.remove_task(self.task) + self.task = None - self.state_update(live, "Scanning") + self.state_update(live, "Scanning") + + def run(self): + if not self._client.comms_check(): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + + if self._single_diff: + self.gateway_diff_load() + + with Live(self.progress_table(), refresh_per_second=4) as live: + self.run_thread(live) From be53123706c7ec020048280cab61cb9d3e81b555 Mon Sep 17 00:00:00 2001 From: Aeyohan Furtado Date: Tue, 14 Jul 2026 11:02:20 +1000 Subject: [PATCH 4/6] tools: ota_upgrade: manually specify client Update relevant functions to manually specify which `LocalClient` is being used. This will allow relevant functions to be called to a specific gateway client. Signed-off-by: Aeyohan Furtado --- src/infuse_iot/tools/ota_upgrade.py | 36 ++++++++++++++++------------- 1 file changed, 20 insertions(+), 16 deletions(-) diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index ae01ce4..3bc6012 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -159,13 +159,13 @@ def data_progress_cb(self, offset): self.task = self.progress.add_task("", total=len(self.patch_file)) self.progress.update(self.task, completed=offset) - def gateway_diff_load(self): + def gateway_diff_load(self, client: LocalClient): assert self._single_diff is not None with self._single_diff.open("rb") as f: patch_file = f.read() - with self._client.connection(InfuseID.GATEWAY, GatewayRequestConnectionRequest.DataType.COMMAND, 10) as _mtu: - rpc_client = RpcClient(self._client, _mtu, InfuseID.GATEWAY) + with client.connection(InfuseID.GATEWAY, GatewayRequestConnectionRequest.DataType.COMMAND, 10) as _mtu: + rpc_client = RpcClient(client, _mtu, InfuseID.GATEWAY) params = file_write_basic.request(rpc_enum_file_action.FILE_FOR_COPY, binascii.crc32(patch_file)) print(f"Writing '{self._single_diff}' to gateway") @@ -179,12 +179,12 @@ def gateway_diff_load(self): ) return_code = hdr.return_code if hdr else -1 if return_code != 0: - sys.exit(f"Failed to save diff file to gateway (({errno.strerror(-return_code)}))") + raise RuntimeError(f"Failed to save diff file to gateway (({errno.strerror(-return_code)}))") print(f"'{self._single_diff}' written to gateway") - def run_file_upload(self, live: Live, mtu: int, source: HopReceived): + def run_file_upload(self, live: Live, mtu: int, source: HopReceived, client: LocalClient): self.state_update(live, f"Uploading patch file to {source.infuse_id:016X}") - rpc_client = RpcClient(self._client, mtu, source.infuse_id) + rpc_client = RpcClient(client, mtu, source.infuse_id) params = file_write_basic.request(rpc_enum_file_action.APP_CPATCH, binascii.crc32(self.patch_file)) @@ -202,9 +202,9 @@ def run_file_upload(self, live: Live, mtu: int, source: HopReceived): elif hdr.return_code == 0: self._pending[source.infuse_id] = time.time() + 60 - def run_file_copy(self, live: Live, mtu: int, source: HopReceived): + def run_file_copy(self, live: Live, mtu: int, source: HopReceived, client: LocalClient): self.state_update(live, f"Copying patch file to {source.infuse_id:016X}") - rpc_client = RpcClient(self._client, mtu, InfuseID.GATEWAY) + rpc_client = RpcClient(client, mtu, InfuseID.GATEWAY) params = bt_file_copy_basic.request( source.interface_address.val.to_rpc_struct(), @@ -227,11 +227,12 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived): elif hdr.return_code == 0: self._pending[source.infuse_id] = time.time() + 60 elif hdr.return_code < 0: + sock_name = client._input_sock.getsockname() err = errno.strerror(-hdr.return_code) - print(f"Failed to copy patch file to {source.infuse_id:016X} ({err})") + print(f"{sock_name} Failed to copy patch file to {source.infuse_id:016X} ({err})") - def run_thread(self, live: Live): - for source, announce in self._client.observe_announce(): + def run_thread(self, live: Live, client: LocalClient): + for source, announce in client.observe_announce(): self.state_update(live, "Scanning") if len(self._explicit_ids): if source.infuse_id not in self._explicit_ids: @@ -314,13 +315,13 @@ def run_thread(self, live: Live): # Attempt to upload self.state_update(live, f"Connecting to {source.infuse_id:016X}") try: - with self._client.connection( + with client.connection( source.infuse_id, GatewayRequestConnectionRequest.DataType.COMMAND, self._conn_timeout ) as mtu: if self._single_diff: - self.run_file_copy(live, mtu, source) + self.run_file_copy(live, mtu, source, client) else: - self.run_file_upload(live, mtu, source) + self.run_file_upload(live, mtu, source, client) except ConnectionRefusedError: self.state_update(live, "Scanning") @@ -338,7 +339,10 @@ def run(self): sys.exit("No communications gateway detected (infuse gateway/bt_native)") if self._single_diff: - self.gateway_diff_load() + try: + self.gateway_diff_load(self._client) + except RuntimeError as e: + sys.exit(str(e)) with Live(self.progress_table(), refresh_per_second=4) as live: - self.run_thread(live) + self.run_thread(live, self._client) From c24e14dde140f7a6b4267053d99e25d5fa9f461d Mon Sep 17 00:00:00 2001 From: Aeyohan Furtado Date: Tue, 14 Jul 2026 12:00:57 +1000 Subject: [PATCH 5/6] tools: ota_upgrade: UI and state handling. Updated multiple state variables to accommodate handling multi-device state. Moved connection, copying and uploading to their own state variables. Updated table UI to use new state variables. Updated UI redraws to not depend on state change. Signed-off-by: Aeyohan Furtado --- src/infuse_iot/tools/ota_upgrade.py | 156 +++++++++++++++++----------- 1 file changed, 94 insertions(+), 62 deletions(-) diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index 3bc6012..e1ca716 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -14,6 +14,7 @@ from rich.progress import ( DownloadColumn, Progress, + TaskID, TransferSpeedColumn, ) from rich.status import Status @@ -83,6 +84,10 @@ def __init__(self, args): self._app_id = app_meta["id"] self._new_ver = app_meta["version"] self._board_crc = crc16_ccitt(app_meta["board"].encode("utf-8")) + self._state_connecting: set[int] = set() + self._state_copying: set[int] = set() + self._state_uploading: set[int] = set() + self._tasks: dict[LocalClient, TaskID] = {} self._handled: list[int] = [] self._pending: dict[int, float] = {} self._missing_diffs: set[str] = set() @@ -97,7 +102,6 @@ def __init__(self, args): DownloadColumn(), TransferSpeedColumn(), ) - self.task = None if args.log is None: self._log = None else: @@ -127,6 +131,10 @@ def add_parser(cls, parser): add_server_port_parser(parser) + @property + def _actioning(self) -> set[int]: + return self._state_connecting | self._state_copying | self._state_uploading + def progress_table(self): table = Table() table.add_column(f"{self._app_name}\n{self._new_ver}") @@ -144,7 +152,14 @@ def progress_table(self): meta = Table(box=None) meta.add_column() meta.add_row(table) - meta.add_row(Status(self.state)) + if self._state_connecting: + meta.add_row(Status(f"Connecting to: {', '.join(f'{i:016X}' for i in self._state_connecting)}")) + processing = self._state_copying | self._state_uploading + if processing: + meta.add_row(Status(f"Writing patch file: {', '.join(f'{i:016X}' for i in processing)}")) + + if not (self._state_connecting or self._state_copying or self._state_uploading): + meta.add_row(Status(self.state)) meta.add_row(self.progress) return meta @@ -153,11 +168,12 @@ def state_update(self, live: Live, state: str): self.state = state live.update(self.progress_table()) - def data_progress_cb(self, offset): - if self.task is None: - self.state = "Writing patch file" - self.task = self.progress.add_task("", total=len(self.patch_file)) - self.progress.update(self.task, completed=offset) + def data_progress_cb(self, offset, client: LocalClient): + task = self._tasks.get(client) + if task is None: + task = self.progress.add_task("", total=len(self.patch_file)) + self._tasks[client] = task + self.progress.update(task, completed=offset) def gateway_diff_load(self, client: LocalClient): assert self._single_diff is not None @@ -183,57 +199,68 @@ def gateway_diff_load(self, client: LocalClient): print(f"'{self._single_diff}' written to gateway") def run_file_upload(self, live: Live, mtu: int, source: HopReceived, client: LocalClient): - self.state_update(live, f"Uploading patch file to {source.infuse_id:016X}") - rpc_client = RpcClient(client, mtu, source.infuse_id) - - params = file_write_basic.request(rpc_enum_file_action.APP_CPATCH, binascii.crc32(self.patch_file)) - - hdr, _rsp = rpc_client.run_data_send_cmd( - file_write_basic.COMMAND_ID, - Auth.DEVICE, - bytes(params), - self.patch_file, - self.data_progress_cb, - file_write_basic.response.from_buffer_copy, - ) + try: + self._state_uploading.add(source.infuse_id) + live.update(self.progress_table()) + # self.state_update(live, f"Uploading patch file to {source.infuse_id:016X}") + rpc_client = RpcClient(client, mtu, source.infuse_id) + + params = file_write_basic.request(rpc_enum_file_action.APP_CPATCH, binascii.crc32(self.patch_file)) + + hdr, _rsp = rpc_client.run_data_send_cmd( + file_write_basic.COMMAND_ID, + Auth.DEVICE, + bytes(params), + self.patch_file, + lambda offset: self.data_progress_cb(offset, client), + file_write_basic.response.from_buffer_copy, + ) - if hdr is None: - self._failed += 1 - elif hdr.return_code == 0: - self._pending[source.infuse_id] = time.time() + 60 + if hdr is None: + self._failed += 1 + elif hdr.return_code == 0: + self._pending[source.infuse_id] = time.time() + 60 + finally: + self._state_uploading.remove(source.infuse_id) + live.update(self.progress_table()) def run_file_copy(self, live: Live, mtu: int, source: HopReceived, client: LocalClient): - self.state_update(live, f"Copying patch file to {source.infuse_id:016X}") - rpc_client = RpcClient(client, mtu, InfuseID.GATEWAY) - - params = bt_file_copy_basic.request( - source.interface_address.val.to_rpc_struct(), - rpc_enum_file_action.APP_CPATCH, - 0, - len(self.patch_file), - binascii.crc32(self.patch_file), - 1, - 3, - ) + try: + self._state_uploading.add(source.infuse_id) + live.update(self.progress_table()) + rpc_client = RpcClient(client, mtu, InfuseID.GATEWAY) + + params = bt_file_copy_basic.request( + source.interface_address.val.to_rpc_struct(), + rpc_enum_file_action.APP_CPATCH, + 0, + len(self.patch_file), + binascii.crc32(self.patch_file), + 1, + 3, + ) - hdr, _rsp = rpc_client.run_standard_cmd( - bt_file_copy_basic.COMMAND_ID, - Auth.DEVICE, - bytes(params), - bt_file_copy_basic.response.from_buffer_copy, - ) - if hdr is None: - self._failed += 1 - elif hdr.return_code == 0: - self._pending[source.infuse_id] = time.time() + 60 - elif hdr.return_code < 0: - sock_name = client._input_sock.getsockname() - err = errno.strerror(-hdr.return_code) - print(f"{sock_name} Failed to copy patch file to {source.infuse_id:016X} ({err})") + hdr, _rsp = rpc_client.run_standard_cmd( + bt_file_copy_basic.COMMAND_ID, + Auth.DEVICE, + bytes(params), + bt_file_copy_basic.response.from_buffer_copy, + ) + if hdr is None: + self._failed += 1 + elif hdr.return_code == 0: + self._pending[source.infuse_id] = time.time() + 60 + elif hdr.return_code < 0: + sock_name = client._input_sock.getsockname() + err = errno.strerror(-hdr.return_code) + print(f"{sock_name} Failed to copy patch file to {source.infuse_id:016X} ({err})") + finally: + self._state_uploading.remove(source.infuse_id) + live.update(self.progress_table()) def run_thread(self, live: Live, client: LocalClient): for source, announce in client.observe_announce(): - self.state_update(live, "Scanning") + live.update(self.progress_table()) if len(self._explicit_ids): if source.infuse_id not in self._explicit_ids: continue @@ -275,7 +302,7 @@ def run_thread(self, live: Live, client: LocalClient): if v_str == self._new_ver and announce.application == self._app_id: self._handled.append(source.infuse_id) self._already += 1 - self.state_update(live, "Scanning") + live.update(self.progress_table()) if self._log: self._log.write( f"{time.time()},0x{source.infuse_id:016x},0x{self._app_id:08x},{v_str},already\n" @@ -293,7 +320,7 @@ def run_thread(self, live: Live, client: LocalClient): self._missing_diffs.add(v_str) self._handled.append(source.infuse_id) self._no_diff += 1 - self.state_update(live, "Scanning") + live.update(self.progress_table()) continue if self._single_diff and self._single_diff != diff_file: @@ -301,7 +328,7 @@ def run_thread(self, live: Live, client: LocalClient): self._missing_diffs.add(v_str) self._handled.append(source.infuse_id) self._no_diff += 1 - self.state_update(live, "Scanning") + live.update(self.progress_table()) continue # Is signal strong enough to connect? @@ -312,27 +339,32 @@ def run_thread(self, live: Live, client: LocalClient): with open(diff_file, "rb") as f: self.patch_file = f.read() + if source.infuse_id in self._actioning: + continue + # Attempt to upload - self.state_update(live, f"Connecting to {source.infuse_id:016X}") + self._state_connecting.add(source.infuse_id) + live.update(self.progress_table()) try: with client.connection( source.infuse_id, GatewayRequestConnectionRequest.DataType.COMMAND, self._conn_timeout ) as mtu: + self._state_connecting.remove(source.infuse_id) if self._single_diff: self.run_file_copy(live, mtu, source, client) else: self.run_file_upload(live, mtu, source, client) except ConnectionRefusedError: - self.state_update(live, "Scanning") + self._state_connecting.remove(source.infuse_id) except ConnectionAbortedError: - self.state_update(live, "Scanning") - - if self.task is not None: - self.progress.remove_task(self.task) - self.task = None + if source.infuse_id in self._state_connecting: + self._state_connecting.remove(source.infuse_id) - self.state_update(live, "Scanning") + if client in self._tasks: + self.progress.remove_task(self._tasks[client]) + del self._tasks[client] + live.update(self.progress_table()) def run(self): if not self._client.comms_check(): From 592847616b04f0d03e2b4c0b0e19ee4ade5c5c2a Mon Sep 17 00:00:00 2001 From: Aeyohan Furtado Date: Tue, 14 Jul 2026 12:25:39 +1000 Subject: [PATCH 6/6] tools: ota_upgrade: support multiple server ports Enable mutliple gateway server ports. Each gateway is handled by a separate thread. UI can handle displaying status on multiple gateways. Signed-off-by: Aeyohan Furtado --- src/infuse_iot/tools/ota_upgrade.py | 70 +++++++++++++++++++++++++---- 1 file changed, 61 insertions(+), 9 deletions(-) diff --git a/src/infuse_iot/tools/ota_upgrade.py b/src/infuse_iot/tools/ota_upgrade.py index e1ca716..419fbf7 100644 --- a/src/infuse_iot/tools/ota_upgrade.py +++ b/src/infuse_iot/tools/ota_upgrade.py @@ -8,6 +8,7 @@ import argparse import binascii import sys +import threading import time from rich.live import Live @@ -41,7 +42,7 @@ class SubCommand(InfuseCommand): DESCRIPTION = "Automatically OTA upgrade observed devices" def __init__(self, args): - self._client = LocalClient(args.server_sock, 1.0) + self._clients = [LocalClient(addr, 1.0) for addr in args.server_sock] self._conn_timeout = args.conn_timeout self._min_rssi: int | None = args.rssi self._explicit_ids: list[int] = [] @@ -113,6 +114,7 @@ def __init__(self, args): with args.list.open("r") as f: for line in f.readlines(): self._explicit_ids.append(int(line.strip(), 0)) + self.end = False @classmethod def add_parser(cls, parser): @@ -129,7 +131,7 @@ def add_parser(cls, parser): explicit.add_argument("--id", type=InfuseDeviceId, help="Single device to upgrade") explicit.add_argument("--list", type=ValidFile, help="File containing a list of IDs to upgrade") - add_server_port_parser(parser) + add_server_port_parser(parser, multi_port=True) @property def _actioning(self) -> set[int]: @@ -260,6 +262,9 @@ def run_file_copy(self, live: Live, mtu: int, source: HopReceived, client: Local def run_thread(self, live: Live, client: LocalClient): for source, announce in client.observe_announce(): + if self.end: + return + live.update(self.progress_table()) if len(self._explicit_ids): if source.infuse_id not in self._explicit_ids: @@ -367,14 +372,61 @@ def run_thread(self, live: Live, client: LocalClient): live.update(self.progress_table()) def run(self): - if not self._client.comms_check(): - sys.exit("No communications gateway detected (infuse gateway/bt_native)") + # Check Gateways are available + unavailable: list[LocalClient] = [] + for client in self._clients: + if not client.comms_check(): + unavailable.append(client) + if len(unavailable) != 0: + print( + f"Warning: Could not use {len(unavailable)} gateways on port(s)" + f" {[x._input_sock.getsockname() for x in unavailable]}." + ) + # If requested, load single diff onto gateways. if self._single_diff: - try: - self.gateway_diff_load(self._client) - except RuntimeError as e: - sys.exit(str(e)) + for client in unavailable: + self._clients.remove(client) + for client in self._clients: + try: + self.gateway_diff_load(client) + except RuntimeError as e: + unavailable.append(client) + port_name = client._input_sock.getsockname() + print(f"Skipping Gateway on port {port_name}: {''.join(e.args)}.") + + # Ensure there is at least one operational gateway available + if len(unavailable) == len(self._clients): + sys.exit("No communications gateway detected (infuse gateway/bt_native)") + for client in unavailable: + self._clients.remove(client) + + if len(self._clients) > 1: + print( + f"running on {len(self._clients)} gateways " + f"{[x._input_sock.getsockname() for x in self._clients]}" + ) + threads: list[threading.Thread] = [] with Live(self.progress_table(), refresh_per_second=4) as live: - self.run_thread(live, self._client) + for client in self._clients: + socket = client._input_sock.getsockname() + t = threading.Thread( + target=self.run_thread, + args=(live, client), + name=f"OTA Upgrade {socket}", + ) + threads.append(t) + if len(threads) > 1: + try: + for t in threads: + t.start() + for t in threads: + t.join() + except KeyboardInterrupt: + self.end = True + self.state_update(live, "Shutting down...") + for t in threads: + t.join() + else: + threads[0].run()