feat(martin-cp): indicativ based progress bar

Cargo.toml

@@ -56,6 +56,7 @@ flate2 = "1"
 flume = "0.12"
 futures = "0.3"
 image = "0.25.8"
+indicatif = "0.18.3"
 indoc = "2"
 insta = "1.44.3"
 itertools = "0.14"
@@ -104,6 +105,7 @@ tilejson = "0.4"
 tokio = { version = "1", features = ["macros"] }
 tokio-postgres-rustls = "0.13"
 tracing = "0.1.44"
+tracing-indicatif = "0.3"
 tracing-log = { version = "0.2.0", features = ["interest-cache"] }
 tracing-subscriber = { version = "0.3", default-features = false }
 tracing-test = "0.2"

martin/Cargo.toml

@@ -132,6 +132,7 @@ dashmap.workspace = true
 enum-display = { workspace = true, optional = true }
 futures.workspace = true
 image = { workspace = true, optional = true }
+indicatif.workspace = true
 itertools.workspace = true
 json-patch = { workspace = true, optional = true }
 lambda-web = { workspace = true, optional = true }
@@ -154,6 +155,7 @@ thiserror.workspace = true
 tilejson.workspace = true
 tokio = { workspace = true, features = ["io-std"] }
 tracing.workspace = true
+tracing-indicatif.workspace = true
 tracing-log = { workspace = true, features = ["interest-cache"] }
 tracing-subscriber = { workspace = true, default-features = false, features = ["env-filter", "json", "fmt", "std", "ansi", "chrono"] }
 url.workspace = true

martin/src/bin/martin-cp.rs

@@ -1,6 +1,6 @@
 use std::borrow::Cow;
 use std::env;
-use std::fmt::{Debug, Display, Formatter};
+use std::fmt::{Debug, Formatter};
 use std::num::NonZeroUsize;
 use std::ops::RangeInclusive;
 use std::path::PathBuf;
@@ -15,6 +15,7 @@ use clap::builder::Styles;
 use clap::builder::styling::AnsiColor;
 use futures::TryStreamExt;
 use futures::stream::{self, StreamExt};
+use indicatif::{ProgressBar, ProgressStyle};
 use martin::config::args::{Args, ExtraArgs, MetaArgs, SrvArgs};
 use martin::config::file::{Config, ServerState, read_config};
 use martin::logging::{ensure_martin_core_log_level_matches, init_tracing};
@@ -257,23 +258,48 @@ impl Debug for TileXyz {
 }
 
 struct Progress {
-    // needed to compute elapsed time
-    start_time: Instant,
-    total: u64,
+    bar: ProgressBar,
     empty: AtomicU64,
     non_empty: AtomicU64,
 }
 
 impl Progress {
     pub fn new(tiles: &[TileRect]) -> Self {
         let total = tiles.iter().map(TileRect::size).sum();
+        let bar = ProgressBar::new(total);
+        bar.set_style(
+            ProgressStyle::default_bar()
+                .template("{elapsed_precise} -> eta: {eta} [{bar:40.cyan/blue} {percent}%] {pos}/{human_len} ({per_sec}) | {msg}")
+                .expect("Invalid progress bar template")
+                .progress_chars("█▓▒░ "),
+        );
         Progress {
-            start_time: Instant::now(),
-            total,
+            bar,
             empty: AtomicU64::default(),
             non_empty: AtomicU64::default(),
         }
     }
+
+    fn update_message(&self) {
+        let non_empty = self.non_empty.load(Ordering::Relaxed);
+        let empty = self.empty.load(Ordering::Relaxed);
+        self.bar.set_message(format!("✓ {non_empty} □ {empty}"));
+    }
+
+    fn inc_empty(&self) {
+        self.empty.fetch_add(1, Ordering::Relaxed);
+        self.bar.inc(1);
+    }
+
+    fn inc_non_empty(&self) {
+        self.non_empty.fetch_add(1, Ordering::Relaxed);
+        self.bar.inc(1);
+    }
+
+    fn finish(&self) {
+        self.update_message();
+        self.bar.finish();
+    }
 }
 
 type MartinCpResult<T> = Result<T, MartinCpError>;
@@ -300,81 +326,6 @@ enum MartinCpError {
     InvalidBoundingBox(&'static str, Bounds, RangeInclusive<f64>),
 }
 
-fn write_duration(f: &mut impl std::fmt::Write, secs: f32) -> std::fmt::Result {
-    if !secs.is_normal() || secs < 0. {
-        // Nonsense input
-        f.write_str("???")
-    } else if secs < 1. {
-        write!(f, "<1s")
-    } else {
-        // we've already handled cases of inf, or negative
-        #[allow(clippy::cast_possible_truncation)]
-        #[allow(clippy::cast_sign_loss)]
-        let secs = secs.ceil() as u64;
-
-        let (mins, secs) = (secs / 60, secs % 60);
-        let (hrs, mins) = (mins / 60, mins % 60);
-        let (days, hrs) = (hrs / 24, hrs % 24);
-
-        // yes, the order is different. Calc years & days, then calc weeks
-        let (years, days) = (days / 365, days % 365);
-        let (weeks, days) = (days / 7, days % 7);
-
-        if years > 0 {
-            write!(
-                f,
-                "{years}y{weeks:02}w{days:02}d{hrs:02}h{mins:02}m{secs:02}s"
-            )
-        } else if weeks > 0 {
-            write!(f, "{weeks}w{days:02}d{hrs:02}h{mins:02}m{secs:02}s")
-        } else if days > 0 {
-            write!(f, "{days}d{hrs:02}h{mins:02}m{secs:02}s")
-        } else if hrs > 0 {
-            write!(f, "{hrs}h{mins:02}m{secs:02}s")
-        } else if mins > 0 {
-            write!(f, "{mins}m{secs:02}s")
-        } else {
-            write!(f, "{secs}s")
-        }
-    }
-}
-
-impl Display for Progress {
-    #[expect(clippy::cast_precision_loss)]
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        let elapsed = self.start_time.elapsed();
-        let elapsed_s = elapsed.as_secs_f32();
-        let non_empty = self.non_empty.load(Ordering::Relaxed);
-        let empty = self.empty.load(Ordering::Relaxed);
-        let done = non_empty + empty;
-        let percent = done * 100 / self.total;
-        let speed = if elapsed_s > 0.0 {
-            done as f32 / elapsed_s
-        } else {
-            0.0
-        };
-        write!(f, "[")?;
-        write_duration(f, elapsed_s)?;
-
-        write!(
-            f,
-            "] {percent:.2}% @ {speed:.1}/s | ✓ {non_empty} □ {empty}"
-        )?;
-
-        let left = self.total - done;
-        f.write_str(" | ")?;
-        if left == 0 {
-            f.write_str("done")
-        } else if done == 0 {
-            f.write_str("??? left")
-        } else {
-            let secs = elapsed_s * left as f32 / done as f32;
-            write_duration(f, secs)?;
-            f.write_str(" left")
-        }
-    }
-}
-
 /// Given a list of tile ranges, iterate over all tiles in the ranges
 fn iterate_tiles(tiles: Vec<TileRect>) -> impl Iterator<Item = TileCoord> {
     tiles.into_iter().flat_map(|t| {
@@ -483,8 +434,8 @@ async fn run_tile_copy(args: CopyArgs, state: ServerState) -> MartinCpResult<()>
 
     let progress = Progress::new(&tiles);
     info!(
-        "Copying {} {} tiles from {} to {}",
-        progress.total,
+        "Copying {} {} tiles from the source {} to {}",
+        progress.bar.length().unwrap_or(0),
         src.info,
         source_id,
         args.output_file.display()
@@ -515,8 +466,8 @@ async fn run_tile_copy(args: CopyArgs, state: ServerState) -> MartinCpResult<()>
             let mut batch = Vec::with_capacity(BATCH_SIZE);
             while let Some(tile) = rx.recv().await {
                 debug!("Generated tile {tile:?}");
-                let done = if tile.data.is_empty() {
-                    progress.empty.fetch_add(1, Ordering::Relaxed)
+                if tile.data.is_empty() {
+                    progress.inc_empty();
                 } else {
                     batch.push((tile.xyz.z, tile.xyz.x, tile.xyz.y, tile.data));
                     if batch.len() >= BATCH_SIZE || last_saved.elapsed() > SAVE_EVERY {
@@ -526,12 +477,14 @@ async fn run_tile_copy(args: CopyArgs, state: ServerState) -> MartinCpResult<()>
                         batch.clear();
                         last_saved = Instant::now();
                     }
-                    progress.non_empty.fetch_add(1, Ordering::Relaxed)
-                };
+                    progress.inc_non_empty();
+                }
+
+                let done = progress.bar.position();
                 if done % PROGRESS_REPORT_AFTER == (PROGRESS_REPORT_AFTER - 1)
                     && last_reported.elapsed() > PROGRESS_REPORT_EVERY
                 {
-                    info!("{progress}");
+                    progress.update_message();
                     last_reported = Instant::now();
                 }
             }
@@ -544,7 +497,7 @@ async fn run_tile_copy(args: CopyArgs, state: ServerState) -> MartinCpResult<()>
         }
     )?;
 
-    info!("{progress}");
+    progress.finish();
 
     mbt.update_metadata(&mut conn, GrowOnly).await?;
 
@@ -554,7 +507,8 @@ async fn run_tile_copy(args: CopyArgs, state: ServerState) -> MartinCpResult<()>
     }
 
     if !args.skip_agg_tiles_hash {
-        if progress.non_empty.load(Ordering::Relaxed) == 0 {
+        let non_empty_count = progress.non_empty.load(Ordering::Relaxed);
+        if non_empty_count == 0 {
             info!("No tiles were copied, skipping agg_tiles_hash computation");
         } else {
             info!("Computing agg_tiles_hash value...");
@@ -623,7 +577,7 @@ async fn init_schema(
 #[tokio::main]
 async fn main() {
     let filter = ensure_martin_core_log_level_matches(env::var("RUST_LOG").ok(), "martin_cp=");
-    init_tracing(&filter, env::var("RUST_LOG_FORMAT").ok());
+    init_tracing(&filter, env::var("RUST_LOG_FORMAT").ok(), true);
 
     let args = CopierArgs::parse();
     if let Err(e) = start(args).await {

martin/src/bin/martin.rs

@@ -56,7 +56,7 @@ async fn start(args: Args) -> MartinResult<()> {
 #[tokio::main]
 async fn main() {
     let filter = ensure_martin_core_log_level_matches(env::var("RUST_LOG").ok(), "martin=");
-    init_tracing(&filter, env::var("RUST_LOG_FORMAT").ok());
+    init_tracing(&filter, env::var("RUST_LOG_FORMAT").ok(), false);
 
     let args = Args::parse();
     if let Err(e) = start(args).await {