[core] changed console feedback to use rich instead of click

Author: MrClock8163

src/instrumentman/datatransfer/app.py

@@ -2,11 +2,18 @@
 from logging import getLogger
 
 from serial import SerialTimeoutException
-from rich.progress import Progress, TextColumn
-from click_extra import echo
+from rich.progress import Progress, TextColumn, BarColumn, TimeElapsedColumn
 from geocompy.communication import open_serial
 
-from ..utils import echo_green, echo_red, echo_yellow
+from ..utils import (
+    print_success,
+    print_error,
+    print_warning,
+    print_plain,
+    console,
+    theme_progress_interrupted,
+    theme_progress_error
+)
 
 
 def main_download(
@@ -30,39 +37,77 @@ def main_download(
         started = False
         logger.info("Starting data download")
         logger.debug("Waiting for first line...")
-        while True:
-            try:
-                data = com.receive_binary()
-                if not started:
-                    started = True
-                    logger.debug("Received first line...")
-
-                if data == eof_bytes and autoclose and not include_eof:
-                    echo_green("Download finished (end-of-file)")
-                    logger.info("Download finished (end-of-file)")
-                    break
+        with Progress(
+            TextColumn("[progress.description]{task.description}"),
+            BarColumn(),
+            TextColumn("{task.completed} line(s)"),
+            TimeElapsedColumn(),
+            console=console
+        ) as progress:
+            task = progress.add_task("Waiting for data...", total=None)
+
+            lines = 0
+            while True:
+                try:
+                    data = com.receive_binary()
+                    if not started:
+                        started = True
+                        logger.debug("Received first line...")
+                        progress.update(task, description="Receiving data")
+
+                    if data == eof_bytes and autoclose and not include_eof:
+                        logger.info("Download finished (end-of-file)")
+
+                        progress.update(
+                            task,
+                            total=lines,
+                            description="End-Of-File"
+                        )
+                        break
 
-                echo(data.decode("ascii", "replace"))
-                if output is not None:
-                    output.write(data + eol_bytes)
+                    print_plain(data.decode("ascii", "replace"))
+                    lines += 1
+                    progress.update(task, completed=lines)
+                    if output is not None:
+                        output.write(data + eol_bytes)
 
-                if data == eof_bytes and autoclose:
-                    echo_green("Download finished (end-of-file)")
-                    logger.info("Download finished (end-of-file)")
+                    if data == eof_bytes and autoclose:
+                        logger.info("Download finished (end-of-file)")
+                        print_success("Download reached end-of-file")
+
+                        progress.update(
+                            task,
+                            total=lines
+                        )
+                        break
+                except SerialTimeoutException:
+                    if started and autoclose:
+                        logger.info("Download finished (timeout)")
+                        print_success("Download finished due to timeout")
+
+                        progress.update(
+                            task,
+                            total=lines
+                        )
+                        break
+                except KeyboardInterrupt:
+                    logger.info("Download stopped manually")
+                    print_warning("Manually interrupted")
+                    console.push_theme(theme_progress_interrupted)
+                    progress.update(
+                        task,
+                        refresh=True
+                    )
                     break
-            except SerialTimeoutException:
-                if started and autoclose:
-                    echo_green("Download finished (timeout)")
-                    logger.info("Download finished (timeout)")
+                except Exception as e:
+                    print_error(e)
+                    console.push_theme(theme_progress_error)
+                    progress.update(
+                        task,
+                        refresh=True
+                    )
+                    logger.exception("Download interrupted by error")
                     break
-            except KeyboardInterrupt:
-                echo_yellow("Download stopped manually")
-                logger.info("Download stopped manually")
-                break
-            except Exception as e:
-                echo_red(f"Download interrupted by error ({e})")
-                logger.exception("Download interrupted by error")
-                break
 
 
 def main_upload(
@@ -86,18 +131,24 @@ def main_upload(
                 next(file)
 
             with Progress(
-                *Progress.get_default_columns(),
-                TextColumn("{task.completed} line(s)")
+                TextColumn("[progress.description]{task.description}"),
+                BarColumn(),
+                TextColumn("{task.completed} line(s)"),
+                TimeElapsedColumn(),
+                console=console
             ) as progress:
-                for line in progress.track(file, description="Uploading..."):
+                lines = 0
+                task = progress.add_task("Uploading", total=None)
+                for line in file:
+                    lines += 1
                     com.send(line)
+                    progress.update(task, completed=lines)
+
+                progress.update(task, total=lines)
 
         except KeyboardInterrupt:
-            echo_yellow("Upload cancelled")
+            print_warning("Upload cancelled")
             logger.info("Upload cancelled by user")
         except Exception as e:
-            echo_red(f"Upload interrupted by error ({e})")
+            print_error(f"Upload interrupted by error ({e})")
             logger.exception("Upload interrupted by error")
-        else:
-            echo_green("Upload finished")
-            logger.info("Upload finished")

src/instrumentman/filetransfer/app.py

@@ -6,19 +6,18 @@
 from re import compile, IGNORECASE
 from logging import getLogger, Logger
 
-from click_extra import echo
-from rich.console import Console, RenderableType
+from rich.console import 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 rich.progress import Progress, TextColumn, TimeElapsedColumn
 from geocompy.communication import open_serial
 from geocompy.geo import GeoCom
 from geocompy.geo.gctypes import GeoComCode
 from geocompy.geo.gcdata import File, Device
 
-from ..utils import echo_red, echo_green
+from ..utils import print_error, console, theme_progress_error
 
 
 _FILE = {
@@ -203,7 +202,6 @@ def run_listing_tree(
     depth: int = 1
 ) -> None:
     filetype = filetype or "unknown"
-    console = Console(width=120)
     logger.info(
         f"Starting content listing of '{directory}' from '{dev}' device"
     )
@@ -240,6 +238,7 @@ def run_listing_tree(
 
     logger.info("Listing complete")
     treeview = build_file_tree(tree)
+    console.width = 120
     console.print(treeview)
 
 
@@ -271,7 +270,7 @@ def run_download(
         _FILE[filetype]
     )
     if resp_setup.error != GeoComCode.OK or resp_setup.params is None:
-        echo_red("Could not set up file download")
+        print_error("Could not set up file download")
         logger.critical(
             f"Could not set up file download ({resp_setup})"
         )
@@ -280,21 +279,25 @@ def run_download(
     block_count = resp_setup.params
     logger.info(f"Expected number of chunks: {block_count:d}")
 
-    with Progress() as progress:
+    with Progress(
+        *Progress.get_default_columns(),
+        TimeElapsedColumn(),
+        console=console
+    ) as progress:
         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:
+                console.push_theme(theme_progress_error)
+                print_error("An error occured during download")
                 progress.stop()
-                echo_red("An error occured during download")
                 logger.critical(
                     f"An error occured during download ({resp_pull})"
                 )
                 return
 
-            echo(bytes.fromhex(resp_pull.params), file, False)
+            file.write(bytes.fromhex(resp_pull.params))
 
     logger.info("Download complete")
-    echo_green("Download complete")
 
 
 def main_download(

src/instrumentman/inclination/app.py

@@ -7,14 +7,13 @@
 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.geo.gctypes import GeoComCode
 from geocompy.communication import open_serial
 
 from ..calculations import adjust_uniform_single
-from ..utils import echo_green
+from ..utils import print_success, console
 
 
 _LINE = compile(r"^\d+(?:\.\d+)?(?:,\-?\d+\.\d+){2}$")
@@ -137,15 +136,15 @@ def main_merge(
     inputs: list[TextIOWrapper],
     output: TextIOWrapper
 ) -> None:
-    echo("hz_deg,cross_sec,length_sec", output)
+    output.write("hz_deg,cross_sec,length_sec\n")
     for item in inputs:
         for line in item:
             if not _LINE.match(line.strip()):
                 continue
 
-            echo(line, output, False)
+            output.write(line)
 
-    echo_green(f"Merged measurements from {len(inputs)} files.")
+    print_success(f"Merged measurements from {len(inputs)} files.")
 
 
 def main_calc(
@@ -186,24 +185,20 @@ def main_calc(
     direction, inc, _ = Coordinate(x, y, z).to_polar()
 
     if output is None:
-        echo(f"""Axis aligned:
-    inclination X: {inc_x:.1f}" +/- {inc_x_dev:.1f}"
-    inclination Y: {inc_y:.1f}" +/- {inc_y_dev:.1f}"
+        console.print(
+            f"""Axis aligned:
+    inclination X: {inc_x:.1f}" ± {inc_x_dev:.1f}"
+    inclination Y: {inc_y:.1f}" ± {inc_y_dev:.1f}"
 Polar:
     direction: {direction.asunit('deg'):.4f}°
     inclination: {inc.asunit('deg') * 3600:.1f}\""""
-             )
+        )
         return
 
-    echo(
-        "inc_x_sec,inc_x_dev_sec,inc_y_sec,inc_y_dev_sec,dir_deg,inc_sec",
-        output
+    output.write(
+        "inc_x_sec,inc_x_dev_sec,inc_y_sec,inc_y_dev_sec,dir_deg,inc_sec\n"
     )
-
-    echo(
-        (
-            f"{inc_x:.1f},{inc_x_dev:.1f},{inc_y:.1f},{inc_y_dev:.1f},"
-            f"{direction.asunit('deg'):.4f},{inc.asunit('deg') * 3600:.1f}"
-        ),
-        output
+    output.write(
+        f"{inc_x:.1f},{inc_x_dev:.1f},{inc_y:.1f},{inc_y_dev:.1f},"
+        f"{direction.asunit('deg'):.4f},{inc.asunit('deg') * 3600:.1f}\n"
     )

src/instrumentman/jobs/app.py

@@ -6,7 +6,7 @@
 from geocompy.geo import GeoCom
 from geocompy.geo.gctypes import GeoComCode
 
-from ..utils import echo_red, echo_yellow
+from ..utils import print_error, print_warning, console
 
 
 def run_listing(
@@ -16,21 +16,21 @@ def run_listing(
     logger.info("Starting job listing")
     resp_setup = tps.csv.setup_listing()
     if resp_setup.error != GeoComCode.OK:
-        echo_red("Could not set up listing")
+        print_error("Could not set up listing")
         logger.critical(
             f"Could not set up listing ({resp_setup})"
         )
         return
 
     resp_list = tps.csv.list()
     if resp_list.error != GeoComCode.OK or resp_list.params is None:
-        echo_red("Could not start listing")
+        print_error("Could not start listing")
         logger.critical(f"Could not start listing ({resp_list})")
         return
 
     job, file, _, _, _ = resp_list.params
     if job == "" or file == "":
-        echo_yellow("No jobs were found")
+        print_warning("No jobs were found")
         logger.info("No jobs were found")
         return
 
@@ -41,7 +41,7 @@ def run_listing(
         col_file
     )
     table.add_row(job, file)
-    with Live(table):
+    with Live(table, console=console):
         while True:
             resp_list = tps.csv.list()
             if resp_list.error != GeoComCode.OK or resp_list.params is None:

src/instrumentman/morse/app.py

@@ -8,8 +8,8 @@
 from geocompy.communication import open_serial
 
 from ..utils import (
-    echo_red,
-    echo_green
+    print_error,
+    console
 )
 
 
@@ -98,7 +98,8 @@ def relay_message(
     with Progress(
         TextColumn("[progress.description]{task.description}"),
         BarColumn(),
-        TaskProgressColumn()
+        TaskProgressColumn(),
+        console=console
     ) as progress:
         for char in progress.track(encoded, description="Relaying message"):
             match char:
@@ -121,7 +122,6 @@ def relay_message(
                         f"Invalid morse stream character: '{char}'"
                     )
 
-    echo_green("Message complete")
     logger.info("Message complete")
 
 
@@ -142,7 +142,7 @@ def main(
         try:
             message.casefold().encode("ascii")
         except UnicodeEncodeError:
-            echo_red("The message contains non-ASCII characters.")
+            print_error("The message contains non-ASCII characters.")
             logger.critical("Message contains non-ASCII characters.")
             exit(1)