Reduce rich methods

src/anomalib/engine/engine.py

@@ -9,7 +9,7 @@
 from typing import Any
 
 import torch
-from lightning.pytorch.callbacks import Callback, RichModelSummary, RichProgressBar
+from lightning.pytorch.callbacks import Callback
 from lightning.pytorch.loggers import Logger
 from lightning.pytorch.trainer import Trainer
 from lightning.pytorch.utilities.types import _EVALUATE_OUTPUT, _PREDICT_OUTPUT, EVAL_DATALOADERS, TRAIN_DATALOADERS
@@ -407,7 +407,7 @@ def _setup_transform(
 
     def _setup_anomalib_callbacks(self) -> None:
         """Set up callbacks for the trainer."""
-        _callbacks: list[Callback] = [RichProgressBar(), RichModelSummary()]
+        _callbacks: list[Callback] = []
 
         # Add ModelCheckpoint if it is not in the callbacks list.
         has_checkpoint_callback = any(isinstance(c, ModelCheckpoint) for c in self._cache.args["callbacks"])

src/anomalib/models/components/sampling/k_center_greedy.py

@@ -9,9 +9,9 @@
 
 import torch
 from torch.nn import functional as F  # noqa: N812
+from tqdm import tqdm
 
 from anomalib.models.components.dimensionality_reduction import SparseRandomProjection
-from anomalib.utils.rich import safe_track
 
 
 class KCenterGreedy:
@@ -98,7 +98,7 @@ def select_coreset_idxs(self, selected_idxs: list[int] | None = None) -> list[in
 
         selected_coreset_idxs: list[int] = []
         idx = int(torch.randint(high=self.n_observations, size=(1,)).item())
-        for _ in safe_track(sequence=range(self.coreset_size), description="Selecting Coreset Indices."):
+        for _ in tqdm(range(self.coreset_size), desc="Selecting Coreset Indices."):
             self.update_distances(cluster_centers=[idx])
             idx = self.get_new_idx()
             if idx in selected_idxs:

src/anomalib/pipelines/components/base/pipeline.py

@@ -10,7 +10,7 @@
 
 import yaml
 from jsonargparse import ArgumentParser, Namespace
-from rich import print, traceback
+from rich import traceback
 
 from anomalib.utils.logging import redirect_logs
 
@@ -66,9 +66,9 @@ def run(self, args: Namespace | None = None) -> None:
             except Exception:  # noqa: PERF203 catch all exception and allow try-catch in loop
                 logger.exception("An error occurred when running the runner.")
                 print(
-                    f"There were some errors when running [red]{runner.generator.job_class.name}[/red] with"
-                    f" [green]{runner.__class__.__name__}[/green]."
-                    f" Please check [magenta]{log_file}[/magenta] for more details.",
+                    f"There were some errors when running {runner.generator.job_class.name} with"
+                    f" {runner.__class__.__name__}."
+                    f" Please check {log_file} for more details.",
                 )
 
     @staticmethod

src/anomalib/pipelines/components/runners/parallel.py

@@ -8,9 +8,6 @@
 from concurrent.futures import ProcessPoolExecutor
 from typing import TYPE_CHECKING
 
-from rich import print
-from rich.progress import Progress, TaskID
-
 from anomalib.pipelines.components.base import JobGenerator, Runner
 from anomalib.pipelines.types import GATHERED_RESULTS, PREV_STAGE_RESULT
 
@@ -51,15 +48,11 @@ def __init__(self, generator: JobGenerator, n_jobs: int) -> None:
         super().__init__(generator)
         self.n_jobs = n_jobs
         self.processes: dict[int, Future | None] = {}
-        self.progress = Progress()
-        self.task_id: TaskID
         self.results: list[dict] = []
         self.failures = False
 
     def run(self, args: dict, prev_stage_results: PREV_STAGE_RESULT = None) -> GATHERED_RESULTS:
         """Run the job in parallel."""
-        self.task_id = self.progress.add_task(self.generator.job_class.name, total=None)
-        self.progress.start()
         self.processes = dict.fromkeys(range(self.n_jobs))
 
         with ProcessPoolExecutor(max_workers=self.n_jobs, mp_context=multiprocessing.get_context("spawn")) as executor:
@@ -71,12 +64,10 @@ def run(self, args: dict, prev_stage_results: PREV_STAGE_RESULT = None) -> GATHE
                 self.processes[index] = executor.submit(job.run, task_id=index)
             self._await_cleanup_processes(blocking=True)
 
-        self.progress.update(self.task_id, completed=1, total=1)
-        self.progress.stop()
         gathered_result = self.generator.job_class.collect(self.results)
         self.generator.job_class.save(gathered_result)
         if self.failures:
-            msg = f"[bold red]There were some errors with job {self.generator.job_class.name}[/bold red]"
+            msg = f"There were some errors with job {self.generator.job_class.name}"
             print(msg)
             logger.error(msg)
             raise ParallelExecutionError(msg)
@@ -97,4 +88,3 @@ def _await_cleanup_processes(self, blocking: bool = False) -> None:
                     logger.exception("An exception occurred while getting the process result.")
                     self.failures = True
                 self.processes[index] = None
-                self.progress.update(self.task_id, advance=1)

src/anomalib/pipelines/components/runners/serial.py

@@ -5,8 +5,7 @@
 
 import logging
 
-from rich import print
-from rich.progress import track
+from tqdm import tqdm
 
 from anomalib.pipelines.components.base import JobGenerator, Runner
 from anomalib.pipelines.types import GATHERED_RESULTS, PREV_STAGE_RESULT
@@ -29,7 +28,7 @@ def run(self, args: dict, prev_stage_results: PREV_STAGE_RESULT = None) -> GATHE
         results = []
         failures = False
         logger.info(f"Running job {self.generator.job_class.name}")
-        for job in track(self.generator(args, prev_stage_results), description=self.generator.job_class.name):
+        for job in tqdm(self.generator(args, prev_stage_results), desc=self.generator.job_class.name):
             try:
                 results.append(job.run())
             except Exception:  # noqa: PERF203
@@ -38,7 +37,7 @@ def run(self, args: dict, prev_stage_results: PREV_STAGE_RESULT = None) -> GATHE
         gathered_result = self.generator.job_class.collect(results)
         self.generator.job_class.save(gathered_result)
         if failures:
-            msg = f"[bold red]There were some errors with job {self.generator.job_class.name}[/bold red]"
+            msg = f"There were some errors with job {self.generator.job_class.name}"
             print(msg)
             logger.error(msg)
             raise SerialExecutionError(msg)