feat: add configurable logging

smpmgr/common.py

@@ -1,15 +1,18 @@
 """Common CLI helpers from rich, typer, click, etc."""
 import asyncio
+import logging
 from dataclasses import dataclass, fields
 
 import typer
 from rich.progress import Progress, SpinnerColumn, TextColumn
-from serial import SerialException  # type: ignore
+from serial import SerialException
 from smp.exceptions import SMPBadStartDelimiter
 from smpclient import SMPClient
 from smpclient.generics import SMPRequest, TEr0, TEr1, TErr, TRep
 from smpclient.transport.serial import SMPSerialTransport
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class TransportDefinition:
@@ -25,6 +28,9 @@ class Options:
 def get_smpclient(options: Options) -> SMPClient:
     """Return an `SMPClient` 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(), options.transport.port)
     else:
         typer.echo(
@@ -49,10 +55,13 @@ async def connect_with_spinner(smpclient: SMPClient) -> None:
             )
             return
         except asyncio.TimeoutError:
-            typer.echo("Transport error: connection timeout")
+            logger.error("Transport error: connection timeout")
         except SerialException as e:
-            typer.echo(f"Serial transport error: {e.strerror}")
+            logger.error(f"Serial transport error: {e.__class__.__name__} - {e}")
 
+        progress.update(
+            connect_task, description=f"{connect_task_description} error", completed=True
+        )
         raise typer.Exit(code=1)
 
 
@@ -68,15 +77,18 @@ async def smp_request(
         description = description or f"Waiting for response to {request.__class__.__name__}..."
         task = progress.add_task(description=description, total=None)
         try:
-            r = await asyncio.wait_for(
-                smpclient.request(request), timeout=options.timeout  # type: ignore
-            )
+            r = await asyncio.wait_for(smpclient.request(request), timeout=options.timeout)
             progress.update(task, description=f"{description} OK", completed=True)
             return r
         except asyncio.TimeoutError:
             progress.update(task, description=f"{description} timeout", completed=True)
+            logger.error("Timeout waiting for response")
             raise typer.Exit(code=1)
         except SMPBadStartDelimiter:
             progress.update(task, description=f"{description} SMP error", completed=True)
-            typer.echo("Is the device an SMP server?")
+            logger.error("Is the device an SMP server?")
+            raise typer.Exit(code=1)
+        except OSError as e:
+            progress.update(task, description=f"{description} OS error", completed=True)
+            logger.error(f"Connection to device lost: {e.__class__.__name__} - {e}")
             raise typer.Exit(code=1)

smpmgr/image_management.py

@@ -1,6 +1,7 @@
 """The image subcommand group."""
 
 import asyncio
+import logging
 from io import BufferedReader
 from pathlib import Path
 from typing import cast
@@ -25,6 +26,7 @@
 from smpmgr.common import Options, connect_with_spinner, get_smpclient, smp_request
 
 app = typer.Typer(name="image", help="The SMP Image Management Group.")
+logger = logging.getLogger(__name__)
 
 
 @app.command()
@@ -76,9 +78,13 @@ async def upload_with_progress_bar(
         try:
             async for offset in smpclient.upload(image, slot):
                 progress.update(task, completed=offset)
+                logger.info(f"Upload {offset=}")
         except SMPBadStartDelimiter:
             progress.stop()
-            typer.echo("Got an unexpected response, is the device an SMP server?")
+            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)
 
 
@@ -91,9 +97,10 @@ def upload(
     """Upload a FW image."""
 
     try:
-        ImageInfo.load_file(str(file))
-    except Exception as e:
-        typer.echo(f"Inspection of FW image failed: {e}")
+        image_info = ImageInfo.load_file(str(file))
+        logger.info(str(image_info))
+    except Exception:
+        logger.exception("Inspection of FW image failed")
         raise typer.Exit(code=1)
 
     smpclient = get_smpclient(cast(Options, ctx.obj))

smpmgr/logging.py

@@ -0,0 +1,50 @@
+"""Logging configuration for smpmgr."""
+
+import asyncio
+import logging
+from enum import Enum, unique
+from pathlib import Path
+
+import click
+import serial
+import typer
+from rich.logging import RichHandler
+
+
+@unique
+class LogLevel(Enum):
+    CRITICAL = 'CRITICAL'
+    ERROR = 'ERROR'
+    WARNING = 'WARNING'
+    INFO = 'INFO'
+    DEBUG = 'DEBUG'
+    NOTSET = 'NOTSET'
+
+
+def setup_logging(loglevel: LogLevel, logfile: Path | None) -> None:
+    """Setup logging for smpmgr."""
+
+    DEBUG_FORMAT = "%(message)s - %(pathname)s:%(lineno)s"
+    DEFAULT_FORMAT = "%(message)s - %(module)s:%(lineno)s"
+    LOGFILE_FORMAT = "%(asctime)s - %(levelname)s - %(pathname)s:%(lineno)s - %(message)s"
+
+    console_handler = RichHandler(
+        rich_tracebacks=True, tracebacks_suppress=[click, typer, asyncio, serial]
+    )
+    file_handler = logging.FileHandler(logfile) if logfile is not None else None
+
+    logging.basicConfig(
+        level=logging.NOTSET,  # root logger logs everything
+        format=(DEBUG_FORMAT if loglevel == LogLevel.DEBUG else DEFAULT_FORMAT),
+        datefmt="[%X]",
+        handlers=((console_handler,) + ((file_handler,) if file_handler is not None else ())),
+    )
+
+    console_handler.setLevel(loglevel.value)  # UI console log level set from --loglevel
+    logging.info(f"Console log level: {logging.getLevelName(console_handler.level)}")
+
+    logging.info(f"Log file: {logfile}")
+    if file_handler is not None:
+        file_handler.setLevel(logging.DEBUG)  # file logs are always DEBUG
+        file_handler.setFormatter(logging.Formatter(LOGFILE_FORMAT))
+        logging.info(f"Log file {logfile} log level: {logging.getLevelName(file_handler.level)}")

smpmgr/main.py

@@ -1,6 +1,7 @@
 """Entry point for the `smpmgr` application."""
 
 import asyncio
+import logging
 from pathlib import Path
 from typing import cast
 
@@ -22,6 +23,9 @@
     smp_request,
 )
 from smpmgr.image_management import upload_with_progress_bar
+from smpmgr.logging import LogLevel, setup_logging
+
+logger = logging.getLogger(__name__)
 
 app = typer.Typer()
 app.add_typer(os_management.app)
@@ -37,8 +41,13 @@ def options(
     timeout: float = typer.Option(
         2.0, help="Transport timeout in seconds; how long to wait for requests"
     ),
+    loglevel: LogLevel = typer.Option(LogLevel.WARNING.value, help="Debug log level"),
+    logfile: Path = typer.Option(None, help="Log file path"),
 ) -> None:
+    setup_logging(loglevel, logfile)
+
     ctx.obj = Options(timeout=timeout, transport=TransportDefinition(port=port))
+    logger.info(ctx.obj)
 
     # TODO: type of transport is inferred from the argument given (--port, --ble, --usb, etc), but
     # it must be the case that only one is provided.
@@ -54,12 +63,14 @@ def upgrade(
 
     try:
         image_info = ImageInfo.load_file(str(file))
+        logger.info(str(image_info))
     except Exception as e:
         typer.echo(f"Inspection of FW image failed: {e}")
         raise typer.Exit(code=1)
 
     try:
         image_tlv_sha256 = image_info.get_tlv(IMAGE_TLV.SHA256)
+        logger.info(f"IMAGE_TLV_SHA256: {image_tlv_sha256}")
     except TLVNotFound:
         typer.echo("Could not find IMAGE_TLV_SHA256 in image.")
         raise typer.Exit(code=1)