Rich progressbars

src/instrumentman/datatransfer/app.py

@@ -1,7 +1,8 @@
 from io import BufferedWriter, TextIOWrapper
 
 from serial import SerialTimeoutException
-from click_extra import echo, progressbar
+from rich.progress import Progress, TextColumn
+from click_extra import echo
 from geocompy.communication import open_serial
 
 from ..utils import echo_green, echo_red, echo_yellow
@@ -68,15 +69,13 @@ def main_upload(
             for _ in range(skip):
                 next(file)
 
-            count = 0
-            with progressbar(
-                file,
-                label="Uploading data",
-                item_show_func=lambda x: f"{count} line(s)"
-            ) as bar:
-                for line in bar:
+            with Progress(
+                *Progress.get_default_columns(),
+                TextColumn("{task.completed} line(s)")
+            ) as progress:
+                for line in progress.track(file, description="Uploading..."):
                     com.send(line)
-                    count += 1
+
         except Exception as e:
             echo_red(f"Upload interrupted by error ({e})")
             return

src/instrumentman/filetransfer/app.py

@@ -1,17 +1,17 @@
 from __future__ import annotations
 
 from io import BufferedWriter
-from typing import TypedDict, Generator
+from typing import TypedDict, Generator, Callable
 import os
 from re import compile, IGNORECASE
 
-from click._termui_impl import ProgressBar
-from click_extra import echo, progressbar
+from click_extra import echo
 from rich.console import Console, RenderableType
 from rich.text import Text
 from rich.tree import Tree
 from rich.table import Table
 from rich.filesize import decimal
+from rich.progress import Progress, TextColumn
 from geocompy.communication import open_serial
 from geocompy.geo import GeoCom
 from geocompy.geo.gctypes import GeoComCode
@@ -50,7 +50,7 @@ class FileTreeItem(TypedDict):
 
 
 def get_directory_items(
-    bar: ProgressBar[str],
+    updater: Callable[[str], None],
     tps: GeoCom,
     device: str,
     directory: str,
@@ -59,7 +59,7 @@ def get_directory_items(
 ) -> list[FileTreeItem]:
     if depth == 0:
         return []
-    bar.update(1, directory)
+    updater(directory)
     resp_setup = tps.ftr.setup_listing(
         _DEVICE[device],
         _FILE[filetype],
@@ -114,7 +114,7 @@ def get_directory_items(
     tps.ftr.abort_list()
     for item in output:
         item["children"] = get_directory_items(
-            bar,
+            updater,
             tps,
             device,
             f"{directory}/{item['name']}",
@@ -195,11 +195,18 @@ def run_listing_tree(
     filetype: str | None,
     depth: int = 1
 ) -> None:
-    with progressbar(
-        _infinite_iterator(),
-        label="Searching directories",
-        item_show_func=lambda x: "" if x is None else str(x)
-    ) as bar:
+    console = Console(width=120)
+    with Progress(
+        *Progress.get_default_columns(),
+        TextColumn("{task.fields[path]}"),
+        console=console,
+        transient=True
+    ) as progress:
+        task = progress.add_task(
+            "Indexing directories...",
+            total=None,
+            path=""
+        )
         tree: FileTreeItem = {
             "name": (
                 f"{dev.upper()}/{directory}"
@@ -209,18 +216,16 @@ def run_listing_tree(
             "size": 0,
             "date": "unknown",
             "children": get_directory_items(
-                bar,
+                lambda path: progress.update(task, path=path),
                 tps,
                 dev,
                 directory,
                 filetype or "unknown",
                 -1 if depth == 0 else depth
             )
         }
-        bar.finish()
 
     treeview = build_file_tree(tree)
-    console = Console(width=120)
     console.print(treeview)
 
 
@@ -249,14 +254,12 @@ def run_download(
         echo_red(f"Could not set up file download ({resp_setup.error.name})")
         return
 
-    block_count = resp_setup.params
-    with progressbar(
-        range(block_count),
-        label="Downloading"
-    ) as bar:
-        for i in bar:
+    with Progress() as progress:
+        block_count = resp_setup.params
+        for i in progress.track(range(block_count), description="Downloading"):
             resp_pull = download(i + 1)
             if resp_pull.error != GeoComCode.OK or resp_pull.params is None:
+                progress.stop()
                 echo_red(
                     "An error occured during download "
                     f"({resp_setup.error.name})"

src/instrumentman/inclination/app.py

@@ -3,7 +3,10 @@
 from math import tan, atan, degrees
 from re import compile
 
-from click_extra import echo, progressbar
+from rich.console import Console
+from rich.progress import track
+from rich.table import Table, Column
+from click_extra import echo
 from geocompy.data import Angle, Coordinate
 from geocompy.geo import GeoCom
 from geocompy.communication import open_serial
@@ -31,35 +34,53 @@ def run_measure(
         if angles.params is not None:
             start = round(angles.params[0].asunit('deg'))
 
-    echo(
-        "hz_deg,cross_sec,length_sec",
-        output
-    )
-
-    with progressbar(
+    con = Console()
+    values: list[tuple[str, str, str]] = []
+    for a in track(
         range(start, start + cycles * 360, turn),
-        label="Measuring",
-        hidden=output is None
-    ) as bar:
-        for a in bar:
-            hz = Angle(a, 'deg').normalized()
-            tps.aut.turn_to(hz, v)
-
-            sleep(1)
-            fullangles = tps.tmc.get_angle_inclination('MEASURE')
-            if fullangles.params is None:
-                continue
+        description="Measuring",
+        console=con
+    ):
+        hz = Angle(a, 'deg').normalized()
+        tps.aut.turn_to(hz, v)
+
+        sleep(1)
+        fullangles = tps.tmc.get_angle_inclination('MEASURE')
+        if fullangles.params is None:
+            continue
 
-            az = fullangles.params[0]
-            cross = fullangles.params[4]
-            length = fullangles.params[5]
+        az = fullangles.params[0]
+        cross = fullangles.params[4]
+        length = fullangles.params[5]
 
-            echo(
-                f"{az.asunit('deg'):.4f},"
-                f"{cross.asunit('deg') * 3600:.2f},"
+        values.append(
+            (
+                f"{az.asunit('deg'):.4f}",
+                f"{cross.asunit('deg') * 3600:.2f}",
                 f"{length.asunit('deg') * 3600:.2f}",
-                output
             )
+        )
+
+    if output is not None:
+        print(
+            "hz_deg,cross_sec,length_sec",
+            file=output
+        )
+        for line in values:
+            print(
+                ",".join(line),
+                file=output
+            )
+    else:
+        table = Table(
+            Column(r"Hz \[deg]", justify='right'),
+            Column(r"Cross \[sec]", justify='right'),
+            Column(r"Length \[sec]", justify='right')
+        )
+        for line in values:
+            table.add_row(*line)
+
+        con.print(table)
 
 
 def main_measure(

src/instrumentman/morse/app.py

@@ -1,7 +1,7 @@
 from time import sleep
 from typing import Callable, Any
 
-from click_extra import progressbar
+from rich.progress import Progress, TextColumn, BarColumn, TaskProgressColumn
 
 from geocompy.geo import GeoCom
 from geocompy.communication import open_serial
@@ -90,12 +90,12 @@ def relay_message(
     unittime: float
 ) -> None:
     encoded = encode_message(message)
-    with progressbar(
-        encoded,
-        label="Relaying message",
-        show_eta=False
-    ) as stream:
-        for char in stream:
+    with Progress(
+        TextColumn("[progress.description]{task.description}"),
+        BarColumn(),
+        TaskProgressColumn()
+    ) as progress:
+        for char in progress.track(encoded, description="Relaying message"):
             match char:
                 case ".":
                     beepstart()