|
20 | 20 | from smpclient import SMPClient
|
21 | 21 | from smpclient.generics import error, success
|
22 | 22 | from smpclient.mcuboot import ImageInfo
|
23 |
| -from smpclient.requests.image_management import ImageStatesRead |
| 23 | +from smpclient.requests.image_management import ImageStatesRead, ImageStatesWrite |
24 | 24 | from typing_extensions import Annotated
|
25 | 25 |
|
26 | 26 | from smpmgr.common import Options, connect_with_spinner, get_smpclient, smp_request
|
@@ -56,6 +56,33 @@ async def f() -> None:
|
56 | 56 | asyncio.run(f())
|
57 | 57 |
|
58 | 58 |
|
| 59 | +@app.command() |
| 60 | +def state_write( |
| 61 | + ctx: typer.Context, |
| 62 | + hash: Annotated[str, typer.Argument(help="SHA256 hash of the image header and body.")], |
| 63 | + confirm: Annotated[bool, typer.Argument(help="Confirm the image given by hash.")], |
| 64 | +) -> None: |
| 65 | + """Request to write the state of FW images on the SMP Server.""" |
| 66 | + |
| 67 | + options = cast(Options, ctx.obj) |
| 68 | + smpclient = get_smpclient(options) |
| 69 | + hash_bytes = bytes.fromhex(hash) |
| 70 | + |
| 71 | + async def f() -> None: |
| 72 | + await connect_with_spinner(smpclient, options.timeout) |
| 73 | + |
| 74 | + r = await smp_request(smpclient, options, ImageStatesWrite(hash=hash_bytes, confirm=confirm), "Waiting for image state write...") # type: ignore # noqa |
| 75 | + |
| 76 | + if error(r): |
| 77 | + print(r) |
| 78 | + elif success(r): |
| 79 | + pass |
| 80 | + else: |
| 81 | + raise Exception("Unreachable") |
| 82 | + |
| 83 | + asyncio.run(f()) |
| 84 | + |
| 85 | + |
59 | 86 | async def upload_with_progress_bar(
|
60 | 87 | smpclient: SMPClient, file: typer.FileBinaryRead | BufferedReader, slot: int = 0
|
61 | 88 | ) -> None:
|
|
0 commit comments