This is an automated email from the ASF dual-hosted git repository.
xiaoxiang781216 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nuttx-apps.git
The following commit(s) were added to refs/heads/master by this push:
new 6f8759cab examples/shv-nxboot-updater: update-script TUI impl, bump
pyshv version
6f8759cab is described below
commit 6f8759cabb5facab968f677f4d513e256c195036
Author: Stepan Pressl <[email protected]>
AuthorDate: Fri Dec 12 03:08:49 2025 +0100
examples/shv-nxboot-updater: update-script TUI impl, bump pyshv version
This commit rewrites the shv-nxboot-updater update-script to a tui
based implementation. Also a pyshv version is increased to 0.10.1.
Signed-off-by: Stepan Pressl <[email protected]>
---
examples/shv-nxboot-updater/update-script/gui.py | 236 ---------------------
.../update-script/requirements.txt | 3 +-
.../shv-nxboot-updater/update-script/shvconfirm.py | 65 +++++-
.../shv-nxboot-updater/update-script/shvflasher.py | 157 +++++++++++---
examples/shv-nxboot-updater/update-script/tui.py | 185 ++++++++++++++++
5 files changed, 371 insertions(+), 275 deletions(-)
diff --git a/examples/shv-nxboot-updater/update-script/gui.py
b/examples/shv-nxboot-updater/update-script/gui.py
deleted file mode 100755
index 0298c2c07..000000000
--- a/examples/shv-nxboot-updater/update-script/gui.py
+++ /dev/null
@@ -1,236 +0,0 @@
-#!/usr/bin/env python3
-
-############################################################################
-# apps/examples/shv-nxboot-updater/update-script/gui.py
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership. The
-# ASF licenses this file to you 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.
-#
-############################################################################
-
-import argparse
-import asyncio
-import logging
-import sys
-from typing import Any
-
-from PyQt6.QtCore import Qt, QThread
-from PyQt6.QtWidgets import (
- QApplication,
- QComboBox,
- QDialog,
- QGridLayout,
- QLabel,
- QLineEdit,
- QProgressBar,
- QPushButton,
-)
-from shvconfirm import shv_confirm
-from shvflasher import shv_flasher
-
-log_levels = (
- logging.DEBUG,
- logging.INFO,
- logging.WARNING,
- logging.ERROR,
- logging.CRITICAL,
-)
-
-PROGRESS_STYLE = """
-QProgressBar{
- border: 2px solid grey;
- border-radius: 5px;
- text-align: center
-}
-
-QProgressBar::chunk {
- background-color: green;
-}
-"""
-
-
-def parse_args() -> argparse.Namespace:
- """Parse passed arguments and return result."""
- parser = argparse.ArgumentParser(
- description="GUI application for NuttX firmware flash over SHV"
- )
- parser.add_argument(
- "-v",
- action="count",
- default=0,
- help="Increase verbosity level of logging",
- )
- parser.add_argument(
- "-q",
- action="count",
- default=0,
- help="Decrease verbosity level of logging",
- )
- parser.add_argument(
- "-i",
- "--image",
- dest="image",
- type=str,
- default="nuttx.nximg",
- help="Image path",
- )
- parser.add_argument(
- "-m",
- "--mount",
- dest="target_mount",
- type=str,
- default="test/nuttxdevice",
- help="Target mount location on the SHV broker",
- )
- parser.add_argument(
- "-s",
- "--server",
- dest="shv_server",
- type=str,
- default="tcp://[email protected]:3755?password=xyz",
- help="SHV server/broker",
- )
- return parser.parse_args()
-
-
-async def flashThreadAsync(
- url: str, img: str, path_to_root, progress_bar: QProgressBar
-) -> None:
- queue: asyncio.Queue = asyncio.Queue()
- task = asyncio.create_task(shv_flasher(url, img, path_to_root, queue))
-
- while True:
- progressVal = await queue.get()
- progress_bar.setValue(progressVal)
-
- if progressVal == 100:
- break
-
- await task
-
-
-class flashThread(QThread):
- def __init__(
- self,
- parent: Any,
- url: str,
- img: str,
- path_to_root: str,
- progress_bar: QProgressBar,
- ) -> None:
- QThread.__init__(self, parent)
- self.url = url
- self.img = img
- self.path_to_root = path_to_root
- self.progress_bar = progress_bar
-
- def run(self) -> None:
- asyncio.run(
- flashThreadAsync(self.url, self.img, self.path_to_root,
self.progress_bar)
- )
-
-
-class confirmThread(QThread):
- def __init__(self, parent: Any, url: str, path_to_root: str) -> None:
- QThread.__init__(self, parent)
- self.url = url
- self.path_to_root = path_to_root
-
- def run(self) -> None:
- asyncio.run(shv_confirm(self.url, self.path_to_root))
-
-
-class ShvFlasherGui(QDialog):
- def __init__(self, image=None, target_mount=None, shv_server=None) -> None:
- super().__init__()
- self.setWindowModality(Qt.WindowModality.ApplicationModal)
- self.setWindowTitle("NuttX Firmware Flasher over SHV")
-
- lbl_shv_url = QLabel("SHV RCP URL")
- self.set_interface(shv_server)
-
- lbl_image = QLabel("Image path")
- self.image = QLineEdit(image if image is not None else "", self)
-
- lbl_target_mount = QLabel("Device mount")
- self.target_mount = QLineEdit(
- target_mount if target_mount is not None else "", self
- )
-
- self.progress_bar = QProgressBar(self)
- self.progress_bar.setValue(0)
- self.progress_bar.setStyleSheet(PROGRESS_STYLE)
-
- pb_flash = QPushButton("FLASH")
- pb_confirm = QPushButton("CONFIRM")
- pb_exit = QPushButton("EXIT")
- grid = QGridLayout()
-
- grid.addWidget(lbl_shv_url, 0, 0)
- grid.addWidget(self.interface, 0, 1)
- grid.addWidget(lbl_image, 1, 0)
- grid.addWidget(self.image, 1, 1)
- grid.addWidget(lbl_target_mount, 2, 0)
- grid.addWidget(self.target_mount, 2, 1)
- grid.addWidget(self.progress_bar, 3, 0, 1, 3)
- grid.addWidget(pb_flash, 4, 0)
- grid.addWidget(pb_confirm, 4, 1)
- grid.addWidget(pb_exit, 4, 2)
-
- pb_flash.clicked.connect(self.do_flash)
- pb_confirm.clicked.connect(self.do_confirm)
- pb_exit.clicked.connect(self.reject)
-
- self.setLayout(grid)
-
- def set_interface(self, shv_server=None) -> None:
- self.interface = QComboBox()
- self.interface.setEditable(True)
- # tcp://user@localhost:3755?password=pass
- self.interface.addItems([shv_server if shv_server is not None else ""])
-
- def do_flash(self) -> None:
- rpc_url = str(self.interface.currentText())
- img_path = str(self.image.text())
- path_to_root = str(self.target_mount.text())
- workerFlash = flashThread(
- self, rpc_url, img_path, path_to_root, self.progress_bar
- )
- workerFlash.start()
-
- def do_confirm(self) -> None:
- rpc_url = str(self.interface.currentText())
- path_to_root = str(self.target_mount.text())
- worker = confirmThread(self, rpc_url, path_to_root)
- worker.start()
-
-
-def main() -> None:
- args = parse_args()
- logging.basicConfig(
- level=log_levels[sorted([1 - args.v + args.q, 0, len(log_levels) -
1])[1]],
- format="[%(asctime)s] [%(levelname)s] - %(message)s",
- )
-
- _ = QApplication(sys.argv)
-
- dialog = ShvFlasherGui(
- image=args.image, target_mount=args.target_mount,
shv_server=args.shv_server
- )
- dialog.exec()
-
-
-if __name__ == "__main__":
- main()
diff --git a/examples/shv-nxboot-updater/update-script/requirements.txt
b/examples/shv-nxboot-updater/update-script/requirements.txt
index b2c6ea55c..66dc6feba 100644
--- a/examples/shv-nxboot-updater/update-script/requirements.txt
+++ b/examples/shv-nxboot-updater/update-script/requirements.txt
@@ -1,5 +1,6 @@
-pyshv==0.8.0
+pyshv==0.10.1
PyQt6
argparse
asyncio
logging
+rich
diff --git a/examples/shv-nxboot-updater/update-script/shvconfirm.py
b/examples/shv-nxboot-updater/update-script/shvconfirm.py
index b01743483..af3bebbbd 100644
--- a/examples/shv-nxboot-updater/update-script/shvconfirm.py
+++ b/examples/shv-nxboot-updater/update-script/shvconfirm.py
@@ -18,15 +18,66 @@
#
############################################################################
-from shv import RpcUrl, SHVValueClient
+import asyncio
+from enum import Enum
+from shv.rpcapi.valueclient import SHVValueClient
+from shv.rpcurl import RpcUrl
-async def shv_confirm(connection: str, path_to_root: str) -> None:
- url = RpcUrl.parse(connection)
- client = await SHVValueClient.connect(url)
- assert client is not None
- await client.call(f"{path_to_root}/fwStable", "confirm")
+class ConfirmStep(Enum):
+ INFO = 0
+ END_OK = 1
+ END_ERROR = 2
+
+
+async def _shv_confirm_cleanup(
+ msg: str, queue: asyncio.Queue, client: SHVValueClient | None
+):
+ queue.put_nowait([ConfirmStep.END_ERROR, msg])
+ if client is not None:
+ await client.disconnect()
+
+
+async def shv_confirm(
+ connection: str, path_to_root: str, queue: asyncio.Queue, method_timeout:
float
+):
+ try:
+ url = RpcUrl.parse(connection)
+ except ValueError:
+ await _shv_confirm_cleanup("The supplied URL is invalid!", queue, None)
+ return
+
+ queue.put_nowait([ConfirmStep.INFO, "Connecting to the server"])
+ try:
+ client = await asyncio.wait_for(
+ SHVValueClient.connect(url), timeout=method_timeout
+ )
+ except Exception:
+ await _shv_confirm_cleanup(
+ "Could not connect to the broker. "
+ + "Is the broker running or is the url OK?",
+ queue,
+ None,
+ )
+ return
+ if client is None:
+ await _shv_confirm_cleanup("Could not connect to the broker.", queue,
None)
+ return
+
+ queue.put_nowait([ConfirmStep.INFO, "Confirming the running firmware."])
+ try:
+ await asyncio.wait_for(
+ client.call(f"{path_to_root}/fwStable", "confirm"),
timeout=method_timeout
+ )
+ except TimeoutError:
+ await _shv_confirm_cleanup(
+ "Timeout when calling the confirm method.", queue, client
+ )
+ return
+ except Exception:
+ await _shv_confirm_cleanup("Failed calling the confirm method.",
queue, client)
+ return
- print("FW confirmed.")
await client.disconnect()
+ queue.put_nowait([ConfirmStep.END_OK, 0])
diff --git a/examples/shv-nxboot-updater/update-script/shvflasher.py
b/examples/shv-nxboot-updater/update-script/shvflasher.py
index 8a5be6a96..407d8b5e2 100644
--- a/examples/shv-nxboot-updater/update-script/shvflasher.py
+++ b/examples/shv-nxboot-updater/update-script/shvflasher.py
@@ -21,63 +21,158 @@
import asyncio
import io
import zlib
+from enum import Enum
-from shv import RpcUrl, SHVBytes, SHVValueClient
+from shv import SHVBytes
+from shv.rpcapi.valueclient import SHVValueClient
+from shv.rpcurl import RpcUrl
+
+
+class FlashStep(Enum):
+ INFO = (0,)
+ WRITE_PROGRESS = (1,)
+ WRITE_COMPLETE = (2,)
+ END_OK = (3,)
+ END_ERROR = 4
+
+
+async def _shv_flasher_cleanup(
+ msg: str, queue: asyncio.Queue, client: SHVValueClient | None
+):
+ queue.put_nowait([FlashStep.END_ERROR, msg])
+ if client is not None:
+ await client.disconnect()
async def shv_flasher(
- connection: str, name: str, path_to_root: str, queue: asyncio.Queue | None
-) -> None:
- url = RpcUrl.parse(connection)
- client = await SHVValueClient.connect(url)
- assert client is not None
+ connection: str,
+ name: str,
+ path_to_root: str,
+ queue: asyncio.Queue,
+ method_timeout: float,
+):
+ try:
+ url = RpcUrl.parse(connection)
+ except ValueError:
+ await _shv_flasher_cleanup("The supplied URL is invalid!", queue, None)
+ return
+ queue.put_nowait([FlashStep.INFO, "Connecting to the server"])
+ try:
+ client = await asyncio.wait_for(
+ SHVValueClient.connect(url), timeout=method_timeout
+ )
+ except Exception:
+ await _shv_flasher_cleanup(
+ "Could not connect to the broker. "
+ + "Is the broker running or is the url OK?",
+ queue,
+ None,
+ )
+ return
+ if client is None:
+ await _shv_flasher_cleanup("Could not connect to the broker.", queue,
None)
+ return
+
node_name = f"{path_to_root}/fwUpdate"
node_name_dotdevice = f"{path_to_root}/.device"
-
- res = await client.call(node_name, "stat")
+ try:
+ res = await asyncio.wait_for(
+ client.call(node_name, "stat"), timeout=method_timeout
+ )
+ except TimeoutError:
+ await _shv_flasher_cleanup(
+ "Timeout when calling the stat method.", queue, client
+ )
+ return
+ except Exception:
+ await _shv_flasher_cleanup("Failed calling the stat method.", queue,
client)
+ return
maxfilesize = res[1]
maxwrite = res[5]
- print(f"Received maximum enabled write size {maxwrite}.")
- print(f"Received maximum file size is {maxfilesize}")
- print(f"Started uploading new firmware {name}... this may take some time.")
size = 0
+ queue.put_nowait(
+ [FlashStep.INFO, f"Received maximum enabled write size {maxwrite}."]
+ )
+ queue.put_nowait([FlashStep.INFO, f"Received maximum file size is
{maxfilesize}"])
+ queue.put_nowait([FlashStep.INFO, f"Started uploading new firmware
{name}."])
with open(name, mode="rb") as f:
# first, compute the CRC from the zlib library
# turns out, NuttX uses the same polynomial
- if queue:
- f.seek(0, io.SEEK_END)
- size = f.tell()
- print(f"File's size is {size}")
- f.seek(0, io.SEEK_SET)
- transfers = size / maxwrite
+ f.seek(0, io.SEEK_END)
+ size = f.tell()
+ queue.put_nowait([FlashStep.INFO, f"File's size is {size}"])
+ f.seek(0, io.SEEK_SET)
+ transfers = size / maxwrite
i = 0
crc = 0
while data := f.read(maxwrite):
- print("Uploading chunk n.", i + 1)
crc = zlib.crc32(data, crc)
offset = i * maxwrite
- res = await client.call(node_name, "write", [offset,
SHVBytes(data)])
+ try:
+ res = await asyncio.wait_for(
+ client.call(node_name, "write", [offset, SHVBytes(data)]),
+ timeout=method_timeout,
+ )
+ except TimeoutError:
+ await _shv_flasher_cleanup(
+ "Timeout when calling the write method", queue, client
+ )
+ return
+ except Exception:
+ await _shv_flasher_cleanup(
+ "Failed calling the write method.", queue, client
+ )
+ return
i += 1
- if queue:
- currProgress = (int)((i * 100) / transfers)
- queue.put_nowait(currProgress)
+ currProgress = (int)((i * 100) / transfers)
+ queue.put_nowait([FlashStep.WRITE_PROGRESS, currProgress])
- print("Flashing completed!")
+ queue.put_nowait([FlashStep.WRITE_COMPLETE, "Flashing completed!"])
- # now get the CRC from the device and reset the device, if OK
- res = await client.call(node_name, "crc", [0, size])
+ # Now get the CRC from the device and reset the device, if OK.
+ # The crc can take a while to calculate, so set it to 20 seconds.
+ try:
+ res = await asyncio.wait_for(
+ client.call(node_name, "crc", [0, size]), timeout=20
+ )
+ except TimeoutError:
+ queue.put_nowait([FlashStep.END_ERROR, "Timeout when calling the crc
method."])
+ await client.disconnect()
+ return
+ except Exception:
+ queue.put_nowait([FlashStep.END_ERROR, "Failed calling the crc
method."])
+ await client.disconnect()
+ return
# just to be sure, make it unsigned
res = res & 0xFFFFFFFF
# the result of the CRC is signed, actually, so make reinterpret it as
unsigned
- print(f"Calculated CRC: {hex(crc)} and received: {hex(res)}")
+ queue.put_nowait(
+ [FlashStep.INFO, f"Calculated CRC {hex(crc)} and received {hex(res)}"]
+ )
if res == crc:
- print("Checksum match, perform reset!")
- res = await client.call(node_name_dotdevice, "reset")
- else:
- print("Checksum mismatch!")
+ queue.put_nowait([FlashStep.INFO, "Performing reset on the target
device."])
+ try:
+ res = await asyncio.wait_for(
+ client.call(node_name_dotdevice, "reset"),
timeout=method_timeout
+ )
+ except TimeoutError:
+ queue.put_nowait(
+ [FlashStep.END_ERROR, "Timeout when calling the reset method."]
+ )
+ await client.disconnect()
+ return
+ except Exception:
+ queue.put_nowait([FlashStep.END_ERROR, "Failed calling the reset
method."])
+ await client.disconnect()
+ return
- await client.disconnect()
+ await client.disconnect()
+ queue.put_nowait([FlashStep.END_OK, 0])
+
+ else:
+ await client.disconnect()
+ queue.put_nowait([FlashStep.END_ERROR, "CRC Mismatch!"])
diff --git a/examples/shv-nxboot-updater/update-script/tui.py
b/examples/shv-nxboot-updater/update-script/tui.py
new file mode 100755
index 000000000..6c275cfab
--- /dev/null
+++ b/examples/shv-nxboot-updater/update-script/tui.py
@@ -0,0 +1,185 @@
+#!/usr/bin/env python3
+
+############################################################################
+# apps/examples/shv-nxboot-updater/update-script/gui.py
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you 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.
+#
+############################################################################
+
+import argparse
+import asyncio
+import logging
+import os
+
+import rich
+from rich.progress import Progress
+from shvconfirm import ConfirmStep, shv_confirm
+from shvflasher import FlashStep, shv_flasher
+
+log_levels = (
+ logging.DEBUG,
+ logging.INFO,
+ logging.WARNING,
+ logging.ERROR,
+ logging.CRITICAL,
+)
+
+
+def get_parser():
+ """Parse passed arguments and return result."""
+ parser = argparse.ArgumentParser(description="SHV NXboot Update Python
script")
+ parser.add_argument(
+ "-v",
+ action="count",
+ default=0,
+ help="Increase verbosity level of logging",
+ )
+ parser.add_argument(
+ "-q",
+ action="count",
+ default=0,
+ help="Decrease verbosity level of logging",
+ )
+ parser.add_argument(
+ "-i",
+ "--image",
+ dest="image",
+ type=str,
+ default="nuttx.nximg",
+ help="Image path",
+ )
+ parser.add_argument(
+ "-m",
+ "--mount",
+ dest="target_mount",
+ type=str,
+ default="test/nuttxdevice",
+ help="Target mount location on the SHV broker",
+ )
+ parser.add_argument(
+ "-s",
+ "--server",
+ dest="shv_server",
+ type=str,
+ default="tcp://[email protected]:3755?password=xyz",
+ help="SHV server/broker",
+ )
+ parser.add_argument(
+ "-o",
+ "--operation",
+ type=str,
+ default="flash",
+ help="Requested operation of this script, supported options: flash,
confirm. Default=flash",
+ )
+ return parser
+
+
+async def do_flash(
+ url: str, path_to_img: str, path_to_root: str, method_call_timeout: float
+):
+ queue: asyncio.Queue = asyncio.Queue()
+ asyncio.create_task(
+ shv_flasher(url, path_to_img, path_to_root, queue, method_call_timeout)
+ )
+ progress = None
+ task_id = None
+ while True:
+ val: [FlashStep, int | str] = await queue.get()
+ match val[0]:
+ case FlashStep.INFO:
+ rich.print(f"[cyan]INFO:[/cyan] {str(val[1])}")
+ case FlashStep.WRITE_PROGRESS:
+ if progress is None:
+ progress = Progress()
+ progress.start()
+ task_id = progress.add_task(
+ f"[cyan]Flashing {os.path.basename(path_to_img)}",
+ total=100,
+ completed=0,
+ )
+ progress.update(task_id, completed=int(val[1]))
+ case FlashStep.WRITE_COMPLETE:
+ rich.print("[cyan]INFO:[/cyan] Image sent!")
+ if progress is not None and task_id is not None:
+ progress.stop()
+ progress.remove_task(task_id)
+ progress = None
+ task_id = None
+ case FlashStep.END_OK:
+ if progress is not None and task_id is not None:
+ progress.stop()
+ progress.remove_task(task_id)
+ rich.print("[green]Flashing operation successful![/green]")
+ break
+ case FlashStep.END_ERROR:
+ if progress is not None:
+ progress.stop()
+ progress.remove_task(task_id)
+ rich.print("[red]ERROR:[/red] {str(val[1])}")
+ break
+
+
+async def do_confirm(url: str, path_to_root: str, method_call_timeout: float):
+ queue: asyncio.Queue = asyncio.Queue()
+ task = asyncio.create_task(
+ shv_confirm(url, path_to_root, queue, method_call_timeout)
+ )
+ while True:
+ val: [ConfirmStep, int | str] = await queue.get()
+ match val[0]:
+ case ConfirmStep.INFO:
+ rich.print(f"[cyan]INFO:[/cyan] {str(val[1])}")
+ case ConfirmStep.END_OK:
+ rich.print("[green]Confirm operation successful![/green]")
+ break
+ case ConfirmStep.END_ERROR:
+ rich.print(f"[red]ERROR:[/red] {str(val[1])}")
+ break
+ await task
+
+
+def tui_main() -> None:
+ parser = get_parser()
+ args = parser.parse_args()
+ logging.basicConfig(
+ level=log_levels[sorted([1 - args.v + args.q, 0, len(log_levels) -
1])[1]],
+ format="[%(asctime)s] [%(levelname)s] - %(message)s",
+ )
+ if args.operation != "flash" and args.operation != "confirm":
+ rich.print(f'[red]ERROR:[/red] unsupported "{args.operation}"
operation')
+ parser.print_help()
+ return
+ if args.operation == "flash" and not os.path.isfile(args.image):
+ rich.print(f"[red]ERROR:[/red] {args.image} not found")
+ return
+ # In seconds
+ METHOD_CALL_TIMEOUT = 5
+ match args.operation:
+ case "flash":
+ asyncio.run(
+ do_flash(
+ args.shv_server, args.image, args.target_mount,
METHOD_CALL_TIMEOUT
+ )
+ )
+ case "confirm":
+ asyncio.run(
+ do_confirm(args.shv_server, args.target_mount,
METHOD_CALL_TIMEOUT)
+ )
+
+
+if __name__ == "__main__":
+ tui_main()
