Detecting Non-TTY Output and Disabling Rich Color

#!/usr/bin/env python3
"""
Console/output-mode detection for a Rich-based geospatial CLI.

Usage:
    python clean_output.py ./input --crs EPSG:3857        # interactive
    python clean_output.py ./input --crs EPSG:3857 | cat   # plain, no ANSI
"""
import os
import sys
import time
import argparse
from pathlib import Path
from contextlib import contextmanager

from rich.console import Console
from rich.progress import Progress, BarColumn, TimeRemainingColumn
import rasterio
from rasterio.warp import calculate_default_transform, reproject, Resampling


def _truthy(name: str) -> bool:
    """A variable counts as 'set' if it exists and is not an explicit off value."""
    value = os.environ.get(name)
    if value is None:
        return False
    return value.strip().lower() not in {"", "0", "false", "no"}


def is_interactive(stream=sys.stdout) -> bool:
    """Return True only when the stream is a real terminal AND nothing forces plain.

    NO_COLOR (https://no-color.org) and CI take precedence over TTY detection:
    a developer may run interactively but still want machine-clean output.
    """
    if _truthy("NO_COLOR"):
        return False
    if _truthy("CI"):
        return False
    # isatty() may be absent on some wrapped streams (e.g. pytest capture).
    return bool(getattr(stream, "isatty", lambda: False)())


def make_console(stderr: bool = True) -> Console:
    """Build a Console tuned for the current output destination.

    - Interactive terminal: full color, Rich picks the true width.
    - Piped / redirected / CI: no_color=True and force_terminal=False so
      absolutely no ANSI escape codes reach the pipe.

    Logs go to stderr by default so stdout stays a clean data channel.
    """
    interactive = is_interactive(sys.stdout)
    return Console(
        stderr=stderr,
        # force_terminal=False hard-disables ANSI; None would let Rich guess.
        force_terminal=False if not interactive else None,
        no_color=not interactive,
        # A fixed width keeps wrapped log lines reproducible in CI artifacts.
        width=None if interactive else 100,
        highlight=interactive,
    )


@contextmanager
def batch_progress(console: Console, total: int, label: str):
    """Live Rich bar when interactive; periodic plain log lines otherwise."""
    if is_interactive(sys.stdout):
        with Progress(
            "[progress.description]{task.description}",
            BarColumn(),
            "[progress.percentage]{task.percentage:>3.0f}%",
            TimeRemainingColumn(),
            console=console,
        ) as progress:
            task = progress.add_task(label, total=total)
            yield lambda: progress.advance(task)
    else:
        # Plain mode: no carriage returns, just a line every 10% or 5 seconds.
        state = {"done": 0, "last": time.monotonic()}

        def advance():
            state["done"] += 1
            now = time.monotonic()
            step = max(1, total // 10)
            if state["done"] % step == 0 or now - state["last"] >= 5:
                pct = 100 * state["done"] / total
                console.print(f"{label}: {state['done']}/{total} ({pct:.0f}%)")
                state["last"] = now

        yield advance
        console.print(f"{label}: {total}/{total} (100%) done")


def reproject_one(src_path: Path, dst_path: Path, target_crs: str) -> None:
    with rasterio.open(src_path) as src:
        transform, width, height = calculate_default_transform(
            src.crs, target_crs, src.width, src.height, *src.bounds
        )
        profile = src.profile.copy()
        profile.update(crs=target_crs, transform=transform, width=width, height=height)
        with rasterio.open(dst_path, "w", **profile) as dst:
            for band in range(1, src.count + 1):
                reproject(
                    source=rasterio.band(src, band),
                    destination=rasterio.band(dst, band),
                    src_crs=src.crs,
                    dst_crs=target_crs,
                    resampling=Resampling.bilinear,
                )


def main() -> int:
    parser = argparse.ArgumentParser(description="Reproject a folder of GeoTIFFs")
    parser.add_argument("input_dir", type=Path)
    parser.add_argument("--crs", default="EPSG:4326", help="Target CRS, e.g. EPSG:3857")
    parser.add_argument("--out", type=Path, default=Path("./out"))
    args = parser.parse_args()
    args.out.mkdir(parents=True, exist_ok=True)

    console = make_console(stderr=True)          # logs/progress -> stderr
    result = Console()                           # data payload   -> stdout

    sources = sorted(args.input_dir.glob("*.tif"))
    if not sources:
        console.print("[yellow]No .tif files found[/]")
        return 2

    console.print(f"Reprojecting {len(sources)} rasters to {args.crs}")
    ok = 0
    with batch_progress(console, len(sources), "warp") as advance:
        for src in sources:
            dst = args.out / f"{src.stem}_{args.crs.replace(':', '_')}.tif"
            try:
                reproject_one(src, dst, args.crs)
                ok += 1
            except Exception as exc:                       # noqa: BLE001
                console.print(f"[red]FAIL[/] {src.name}: {exc}")
            advance()

    # Machine-readable summary on stdout, never styled.
    result.print(f"{ok} {len(sources)}")
    return 0 if ok == len(sources) else 12


if __name__ == "__main__":
    sys.exit(main())

# 1. Piped through cat: there must be NO ^[[ sequences in the output.
python clean_output.py ./input --crs EPSG:3857 | cat -v

# 2. Grep for the raw ESC byte (0x1b). Zero matches = clean.
python clean_output.py ./input --crs EPSG:3857 2>&1 | grep -c $'\x1b\['
# expected: 0

# 3. CI simulation: force the plain path even in an interactive shell.
CI=true python clean_output.py ./input --crs EPSG:3857 | cat -v

# 4. Confirm the machine payload on stdout is unstyled and parseable.
python clean_output.py ./input --crs EPSG:3857 2>/dev/null
# expected: two integers, e.g. "12 12", with no color