Sharing Global Options Across Geospatial Subcommands

#!/usr/bin/env python3
"""
Geospatial CLI with global options shared through ctx.obj.

Examples:
    python geotool.py --crs EPSG:32633 --workers 4 raster warp scene.tif out.tif
    python geotool.py --crs EPSG:3857 vector dissolve parcels.gpkg merged.gpkg
"""
from __future__ import annotations

import os
import sys
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

import typer
from pyproj import CRS

app = typer.Typer(no_args_is_help=True, help="Geospatial toolchain with shared globals")
raster_app = typer.Typer(no_args_is_help=True, help="Raster commands")
vector_app = typer.Typer(no_args_is_help=True, help="Vector commands")
app.add_typer(raster_app, name="raster")
app.add_typer(vector_app, name="vector")


@dataclass(frozen=True)
class AppState:
    """Resolved global options shared by every subcommand."""
    crs: str
    workers: int
    verbose: bool


def _load_config(path: Optional[Path]) -> dict:
    """Read a TOML config file, returning an empty dict when absent."""
    if path is None or not path.exists():
        return {}
    with path.open("rb") as fh:
        return tomllib.load(fh).get("defaults", {})


@app.callback()
def main(
    ctx: typer.Context,
    crs: str = typer.Option("EPSG:4326", "--crs", help="Target CRS for all commands"),
    workers: int = typer.Option(1, "--workers", min=1, help="Worker process count"),
    config: Optional[Path] = typer.Option(None, "--config", help="TOML config file"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose logging"),
) -> None:
    """Resolve global options once and stash them on ctx.obj.

    Precedence is defaults, then config file, then environment, then flags.
    Typer has already applied flags over its own defaults, so we only fill
    from the config/env layers when the caller left a value at its default.
    """
    cfg = _load_config(config)
    resolved_crs = crs
    if resolved_crs == "EPSG:4326":  # untouched Typer default
        resolved_crs = os.environ.get("GEOTOOL_CRS", cfg.get("crs", resolved_crs))

    # Validate the CRS at the boundary so no subcommand sees a bad code.
    try:
        CRS.from_user_input(resolved_crs)
    except Exception:
        typer.secho(f"Invalid CRS: {resolved_crs}", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=10)  # 10 = CRS mismatch/invalid

    ctx.obj = AppState(crs=resolved_crs, workers=workers, verbose=verbose)


def get_state(ctx: typer.Context) -> AppState:
    """Fetch shared state, failing loudly if the callback never ran."""
    if ctx.obj is None:
        typer.secho("Global options were not initialised.", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=2)  # 2 = usage error
    return ctx.obj


@raster_app.command("warp")
def raster_warp(
    ctx: typer.Context,
    src: Path = typer.Argument(..., exists=True, help="Source GeoTIFF"),
    dst: Path = typer.Argument(..., help="Destination GeoTIFF"),
    crs: Optional[str] = typer.Option(None, "--crs", help="Override global CRS"),
) -> None:
    """Reproject a raster, reading CRS and workers from the shared state."""
    state = get_state(ctx)
    target_crs = crs or state.crs  # local flag wins, else fall back to global
    if state.verbose:
        typer.echo(f"[raster warp] crs={target_crs} workers={state.workers}")

    import rasterio
    from rasterio.warp import calculate_default_transform, reproject, Resampling

    with rasterio.open(src) as ds:
        transform, width, height = calculate_default_transform(
            ds.crs, target_crs, ds.width, ds.height, *ds.bounds
        )
        profile = ds.profile.copy()
        profile.update(crs=target_crs, transform=transform, width=width, height=height)
        with rasterio.open(dst, "w", **profile) as out:
            for band in range(1, ds.count + 1):
                reproject(
                    source=rasterio.band(ds, band),
                    destination=rasterio.band(out, band),
                    dst_crs=target_crs,
                    resampling=Resampling.bilinear,
                    num_threads=state.workers,  # shared worker budget
                )
    typer.echo(f"Reprojected {src.name} -> {dst.name} in {target_crs}")


@vector_app.command("dissolve")
def vector_dissolve(
    ctx: typer.Context,
    src: Path = typer.Argument(..., exists=True, help="Source vector dataset"),
    dst: Path = typer.Argument(..., help="Destination vector dataset"),
    by: str = typer.Option("region", "--by", help="Attribute to dissolve on"),
) -> None:
    """Dissolve features, reprojecting to the shared global CRS first."""
    state = get_state(ctx)
    if state.verbose:
        typer.echo(f"[vector dissolve] crs={state.crs} workers={state.workers}")

    import geopandas as gpd

    gdf = gpd.read_file(src, engine="pyogrio")  # prefer pyogrio over fiona
    gdf = gdf.to_crs(state.crs)                  # same CRS the raster command uses
    dissolved = gdf.dissolve(by=by)
    dissolved.to_file(dst, engine="pyogrio")
    typer.echo(f"Dissolved {len(gdf)} -> {len(dissolved)} features in {state.crs}")


if __name__ == "__main__":
    app()

# Global flags resolve once; both commands should echo EPSG:32633 and workers=4
python geotool.py --crs EPSG:32633 --workers 4 --verbose \
    raster warp scene.tif scene_utm.tif
python geotool.py --crs EPSG:32633 --workers 4 --verbose \
    vector dissolve parcels.gpkg parcels_merged.gpkg

# Local override wins only for the command it is attached to
python geotool.py --crs EPSG:4326 --verbose \
    raster warp scene.tif scene_web.tif --crs EPSG:3857
# -> [raster warp] crs=EPSG:3857 workers=1

# Missing callback path: a bad global CRS exits 10, not a traceback
python geotool.py --crs EPSG:99999 raster warp scene.tif out.tif; echo "exit=$?"
# -> Invalid CRS: EPSG:99999   /   exit=10