.. module:: CLI

CLI
=========================

Utilities for command-line interfaces, including progress reporting for
long-running pipeline stages.


ProgressTracker
---------------

``ProgressTracker`` provides a TTY-aware progress bar designed for use inside
ETL pipelines and ``ITask`` implementations. It detects ``sys.stdout.isatty()``
once at construction and routes output accordingly, animating on real terminals
and falling back to structured ``logger.info`` calls on Lambda, ECS, or CI
environments where ``\r`` overwriting is not meaningful.

**Why not tqdm?**

``tqdm`` is the de-facto standard for progress bars in Python, but it introduces
an external dependency that is unnecessary for the narrow use case this library
covers. ``core-mixins`` is a foundational utility library, one of its explicit
goals is to remain dependency-free (beyond the ``typing-extensions`` shim for
Python < 3.11). Adding ``tqdm`` would propagate it as a transitive dependency
into every project that depends on ``core-mixins``, including production services
running on AWS Lambda or ECS where package size and supply-chain surface area
both matter.

``ProgressTracker`` deliberately covers only what ETL pipelines stages need:

- A single animated line that updates in place on a real terminal.
- A persistent completion line with a summary note.
- Structured logging as the fallback when there is no terminal.

It does not aim to replace ``tqdm`` in interactive notebooks, download bars,
or multi-bar scenarios. For those use cases, ``tqdm`` remains the right choice.

**Progress bar live animation (TTY)**

.. image:: _static/progress-tracker-animation.png
   :alt: ProgressTracker live animation showing a progress bar at 66.7%

|

**Completion with all stages passed (TTY)**

.. image:: _static/progress-tracker-completion-success.png
   :alt: ProgressTracker completion output — all stages passed, green summary line

|

**Completion with validation errors (TTY)**

.. image:: _static/progress-tracker-completion.png
   :alt: ProgressTracker completion output — validation errors, yellow summary line

|

**CloudWatch output (Lambda / ECS / non-TTY)**

On non-TTY environments ``progress()`` is silent and ``done()`` emits one
structured ``logger.info`` line per stage, no ``\r`` noise in CloudWatch.

.. image:: _static/progress-tracker-cloudwatch-logs.png
   :alt: AWS CloudWatch log events showing one INFO line per pipeline stage

|

**Routing behaviour**

+-------------------------------+--------------------------+-----------------------------+
| Environment                   | ``progress()``           | ``done()``                  |
+===============================+==========================+=============================+
| TTY (real terminal)           | ``\r`` animation         | ``✔`` line to stdout        |
+-------------------------------+--------------------------+-----------------------------+
| Non-TTY (Lambda / ECS / CI)   | silent                   | ``logger.info`` if set      |
+-------------------------------+--------------------------+-----------------------------+

|

**Usage with ITask**

Each pipeline stage is a concrete ``ITask`` subclass. The ``ProgressTracker`` is
instantiated inside ``execute()`` with ``self.logger``, so output routing (TTY
animation vs. CloudWatch logs) is handled automatically.

.. code-block:: python

    import random
    import time

    from core_mixins.cli.progress_tracker import BOLD, GREEN, RESET, YELLOW, ProgressTracker
    from core_mixins.interfaces.task import ITask
    from core_mixins.logger import get_logger


    class IngestTask(ITask):

        def execute(self, *args, **kwargs) -> dict:
            files = 120
            total_mb = 0.0
            t0 = time.monotonic()
            tracker = ProgressTracker(logger=self.logger)

            for i in range(1, files + 1):
                time.sleep(random.uniform(0.01, 0.025))
                total_mb += random.uniform(0.4, 3.2)
                elapsed = time.monotonic() - t0
                pct = i / files * 100
                tracker.progress("Ingesting files", pct, f"{total_mb / elapsed:.1f} MB/s")

            elapsed = time.monotonic() - t0
            tracker.done("Ingesting files", f"{files} files · {total_mb:.0f} MB · {elapsed:.1f}s")
            return {"records": files}


    class ValidateTask(ITask):

        def execute(self, *args, **kwargs) -> dict:
            records = kwargs["records"]
            errors = 0
            t0 = time.monotonic()
            tracker = ProgressTracker(logger=self.logger)

            for i in range(1, records + 1):
                time.sleep(random.uniform(0.003, 0.009))
                if random.random() < 0.015:
                    errors += 1
                pct = i / records * 100
                tracker.progress("Validating records", pct, f"{errors} errors")

            elapsed = time.monotonic() - t0
            tracker.done("Validating records", f"{records} records · {errors} errors · {elapsed:.1f}s")
            return {"records": records, "errors": errors}


    if __name__ == "__main__":
        logger = get_logger(
            __name__,
            formatter="%(message)s",
            reset_handlers=True,
            propagate=False,
        )

        logger.info(f"\n{BOLD}  Pipeline starting…{RESET}\n")
        t0 = time.monotonic()

        ingest   = IngestTask(logger=logger).execute()
        validate = ValidateTask(logger=logger).execute(records=ingest["records"])

        elapsed = time.monotonic() - t0
        errors = validate["errors"]
        verdict = (
            f"{GREEN}{BOLD}  All stages completed in {elapsed:.1f}s.{RESET}"
            if errors == 0
            else f"{YELLOW}{BOLD}  Done in {elapsed:.1f}s — {errors} validation error(s).{RESET}"
        )
        logger.info(f"\n{verdict}\n")


API Reference
-------------

.. automodule:: core_mixins.cli.progress_tracker
   :members:
   :undoc-members:
   :show-inheritance: