Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions smpmgr/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@

import asyncio
import logging
import sys
from dataclasses import dataclass, fields
from typing import Final, Type, TypedDict, TypeVar, assert_never

import typer
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
from serial import SerialException
from smp.exceptions import SMPBadStartDelimiter
Expand Down Expand Up @@ -128,7 +130,9 @@ def get_smpclient(options: 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(
SpinnerColumn(), TextColumn("[progress.description]{task.description}")
SpinnerColumn(spinner_name="line"),
TextColumn("[progress.description]{task.description}"),
console=Console(force_terminal=sys.stdout.isatty()),
) as progress:
connect_task_description = f"Connecting to {smpclient._address}..."
connect_task = progress.add_task(description=connect_task_description, total=None)
Expand Down Expand Up @@ -156,7 +160,9 @@ async def smp_request(
timeout_s: float | None = None,
) -> TRep | TEr1 | TEr2:
with Progress(
SpinnerColumn(), TextColumn("[progress.description]{task.description}")
SpinnerColumn(spinner_name="line"),
TextColumn("[progress.description]{task.description}"),
console=Console(force_terminal=sys.stdout.isatty()),
) as progress:
description = description or f"Waiting for response to {request.__class__.__name__}..."
task = progress.add_task(description=description, total=None)
Expand Down
8 changes: 5 additions & 3 deletions smpmgr/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -265,9 +265,11 @@ async def f() -> None:
image_states_response = await smp_request(
smpclient,
ImageStatesWrite(hash=image_hash, confirm=confirm),
"Marking uploaded image for permanent upgrade..."
if confirm
else "Marking uploaded image for test upgrade...",
(
"Marking uploaded image for permanent upgrade..."
if confirm
else "Marking uploaded image for test upgrade..."
),
)
if success(image_states_response):
pass
Expand Down
Loading