diff --git a/poetry.lock b/poetry.lock index 3104b70..02375d7 100644 --- a/poetry.lock +++ b/poetry.lock @@ -786,6 +786,17 @@ dev = ["autoflake (>=1.3.1,<2.0.0)", "flake8 (>=3.8.3,<4.0.0)", "pre-commit (>=2 doc = ["cairosvg (>=2.5.2,<3.0.0)", "mdx-include (>=1.4.1,<2.0.0)", "mkdocs (>=1.1.2,<2.0.0)", "mkdocs-material (>=8.1.4,<9.0.0)", "pillow (>=9.3.0,<10.0.0)"] test = ["black (>=22.3.0,<23.0.0)", "coverage (>=6.2,<7.0)", "isort (>=5.0.6,<6.0.0)", "mypy (==0.910)", "pytest (>=4.4.0,<8.0.0)", "pytest-cov (>=2.10.0,<5.0.0)", "pytest-sugar (>=0.9.4,<0.10.0)", "pytest-xdist (>=1.32.0,<4.0.0)", "rich (>=10.11.0,<14.0.0)", "shellingham (>=1.3.0,<2.0.0)"] +[[package]] +name = "types-pyserial" +version = "3.5.0.11" +description = "Typing stubs for pyserial" +optional = false +python-versions = ">=3.7" +files = [ + {file = "types-pyserial-3.5.0.11.tar.gz", hash = "sha256:997fcaadfe00386cd80f789a42a5febce4d65774e17e3396d91cac6345b9aab2"}, + {file = "types_pyserial-3.5.0.11-py3-none-any.whl", hash = "sha256:db80b56868b1bcc5667ec25bacb9bf5adaa397daa6225e95466f37370c6bc626"}, +] + [[package]] name = "typing-extensions" version = "4.9.0" @@ -800,4 +811,4 @@ files = [ [metadata] lock-version = "2.0" python-versions = ">=3.10, <3.13" -content-hash = "ba596bd1a2b2c85ee43254f5838800c030b46377d978c85edd219c2ec9766a25" +content-hash = "783422bc14131e88b04c261a87b7140cd8b76c96b969c06c7bec9dcde1e4e87f" diff --git a/pyproject.toml b/pyproject.toml index fb4ce4c..8bd2785 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ isort = "^5.12.0" mypy = "^1.7.0" mypy-extensions = "^1.0.0" pytest-xdist = "^3.4.0" +types-pyserial = "^3.5.0.11" [tool.black] line-length = 100 diff --git a/smpmgr/common.py b/smpmgr/common.py index 17f3eb8..49d5c71 100644 --- a/smpmgr/common.py +++ b/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) diff --git a/smpmgr/image_management.py b/smpmgr/image_management.py index 61585b2..49047f8 100644 --- a/smpmgr/image_management.py +++ b/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)) diff --git a/smpmgr/logging.py b/smpmgr/logging.py new file mode 100644 index 0000000..85135bc --- /dev/null +++ b/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)}") diff --git a/smpmgr/main.py b/smpmgr/main.py index be6aae5..b6ad2d8 100644 --- a/smpmgr/main.py +++ b/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)