Merge pull request #13 from intercreate/feature/intercreate-group

Feature/intercreate group

Author: JPHutchins

poetry.lock

@@ -19,7 +19,7 @@ source = "git-tag"
 
 [tool.poetry.dependencies]
 python = ">=3.10, <3.13"
-smpclient = "^1.0.1"
+smpclient = "^1.3.1"
 typer = {extras = ["all"], version = "^0.9.0"}
 
 

pyproject.toml

@@ -3,6 +3,7 @@
 import asyncio
 import logging
 from dataclasses import dataclass, fields
+from typing import Type, TypeVar
 
 import typer
 from rich.progress import Progress, SpinnerColumn, TextColumn
@@ -14,6 +15,11 @@
 
 logger = logging.getLogger(__name__)
 
+TSMPClient = TypeVar(
+    "TSMPClient",
+    bound=SMPClient,
+)
+
 
 @dataclass(frozen=True)
 class TransportDefinition:
@@ -27,13 +33,13 @@ class Options:
     mtu: int
 
 
-def get_smpclient(options: Options) -> SMPClient:
-    """Return an `SMPClient` to the chosen transport or raise `typer.Exit`."""
+def get_custom_smpclient(options: Options, smp_client_cls: Type[TSMPClient]) -> TSMPClient:
+    """Return an `SMPClient` subclass to the chosen transport or raise `typer.Exit`."""
     if options.transport.port is not None:
         logger.info(
             f"Initializing SMPClient with the SMPSerialTransport, {options.transport.port=}"
         )
-        return SMPClient(SMPSerialTransport(mtu=options.mtu), options.transport.port)
+        return smp_client_cls(SMPSerialTransport(mtu=options.mtu), options.transport.port)
     else:
         typer.echo(
             f"A transport option is required; "
@@ -43,6 +49,11 @@ def get_smpclient(options: Options) -> SMPClient:
         raise typer.Exit(code=1)
 
 
+def get_smpclient(options: Options) -> SMPClient:
+    """Return an `SMPClient` to the chosen transport or raise `typer.Exit`."""
+    return get_custom_smpclient(options, SMPClient)
+
+
 async def connect_with_spinner(smpclient: SMPClient) -> None:
     """Spin while connecting to the SMP Server; raises `typer.Exit` if connection fails."""
     with Progress(

smpmgr/common.py

@@ -25,6 +25,7 @@
 )
 from smpmgr.image_management import upload_with_progress_bar
 from smpmgr.logging import LogLevel, setup_logging
+from smpmgr.user import intercreate
 
 logger = logging.getLogger(__name__)
 
@@ -36,6 +37,7 @@
 app: Final = typer.Typer(help="\n".join(HELP_LINES))
 app.add_typer(os_management.app)
 app.add_typer(image_management.app)
+app.add_typer(intercreate.app)
 
 
 @app.callback(invoke_without_command=True)

smpmgr/main.py

@@ -0,0 +1,79 @@
+"""The Intercreate (ic) subcommand group."""
+
+import asyncio
+import logging
+from io import BufferedReader
+from pathlib import Path
+from typing import cast
+
+import typer
+from rich.progress import (
+    BarColumn,
+    DownloadColumn,
+    Progress,
+    TextColumn,
+    TimeRemainingColumn,
+    TransferSpeedColumn,
+)
+from smp import header as smphdr
+from smp.exceptions import SMPBadStartDelimiter
+from smpclient.extensions import intercreate as ic
+from typing_extensions import Annotated
+
+from smpmgr.common import Options, connect_with_spinner, get_custom_smpclient
+
+app = typer.Typer(
+    name="ic", help=f"The Intercreate User Group ({smphdr.GroupId.INTERCREATE.value})"
+)
+logger = logging.getLogger(__name__)
+
+
+async def upload_with_progress_bar(
+    smpclient: ic.ICUploadClient, file: typer.FileBinaryRead | BufferedReader, image: int = 0
+) -> None:
+    """Animate a progress bar while uploading the data."""
+
+    with Progress(
+        TextColumn("[bold blue]{task.fields[filename]}", justify="right"),
+        BarColumn(),
+        "[progress.percentage]{task.percentage:>3.1f}%",
+        "•",
+        DownloadColumn(),
+        "•",
+        TransferSpeedColumn(),
+        "•",
+        TimeRemainingColumn(),
+    ) as progress:
+        data = file.read()
+        file.close()
+        task = progress.add_task("Uploading", total=len(data), filename=file.name, start=True)
+        try:
+            async for offset in smpclient.ic_upload(data, image):
+                progress.update(task, completed=offset)
+                logger.info(f"Upload {offset=}")
+        except SMPBadStartDelimiter as e:
+            progress.stop()
+            logger.info(f"Bad start delimiter: {e}")
+            logger.error("Got an unexpected response, is the device an SMP server?")
+            raise typer.Exit(code=1)
+        except OSError as e:
+            logger.error(f"Connection to device lost: {e.__class__.__name__} - {e}")
+            raise typer.Exit(code=1)
+
+
+@app.command()
+def upload(
+    ctx: typer.Context,
+    file: Annotated[Path, typer.Argument(help="Path to binary data")],
+    image: Annotated[int, typer.Option(help="The image slot to upload to")] = 0,
+) -> None:
+    """Upload data to custom image slots, like a secondary MCU or external storage."""
+
+    smpclient = get_custom_smpclient(cast(Options, ctx.obj), ic.ICUploadClient)
+
+    async def f() -> None:
+        await connect_with_spinner(smpclient)
+        with open(file, "rb") as f:
+            await upload_with_progress_bar(smpclient, f, image)
+
+    asyncio.run(f())