Rasters, recipes and cli usage

src/globato/cli/build.py

@@ -76,6 +76,16 @@
 @click.option(
     "-W", "--weights", default="auto", help="Weight thresholds ('auto' or '1.0/0.5')."
 )
+@click.option(
+    "--modifier",
+    multiple=True,
+    help="Apply a recipe modifier at runtime (e.g., exclude_module:modules=csb/tnm).",
+)
+@click.option(
+    "--schema",
+    multiple=True,
+    help="Apply a domain schema validation to the recipe.",
+)
 @click.option(
     "--shared-cache",
     # type=click.Path(resolve_path=True),
@@ -87,6 +97,11 @@
 @click.option(
     "--refresh", is_flag=True, help="Force fresh API fetch, bypassing local cache."
 )
+@click.option(
+    "--ignore-failures",
+    is_flag=True,
+    help="Continue processing through failures (Warning: may result in incomplete data or products).",
+)
 @click.argument("sources", nargs=-1)
 def build_cmd(
     region,
@@ -104,11 +119,14 @@ def build_cmd(
     limits,
     weights,
     blend,
+    modifier,
+    schema,
     shared_cache,
     metadata,
     export,
     sources,
     refresh,
+    ignore_failures,
 ):
     """Build a Digital Elevation Model recipe, and execute it."""
 
@@ -128,6 +146,9 @@ def build_cmd(
         res=increment,
     )
 
+    parsed_modifiers = [parse_hook_string(m) for m in modifier]
+    parsed_schemas = [s for s in schema]
+
     base_outdir = os.path.abspath(outdir) if outdir else os.path.abspath(".")
 
     try:
@@ -164,6 +185,7 @@ def build_cmd(
             auto_res_list = [base_res * (3**i) for i in range(len(weight_list) + 1)]
             blend_list = [int_or(b, 10) for b in str(blend).split("/")] if blend else []
 
+        batch_outname = "%name%_%batch_name%"
         # --- Base Pipeline Standard Hooks ---
         global_hooks = [
             {"name": "spatial-crop"},
@@ -173,11 +195,11 @@ def build_cmd(
             {"name": "drop_class"},
             {
                 "name": "provenance",
-                "args": {"res": increment, "output": f"{outname}_provenance.tif"},
+                "args": {"res": increment, "output": f"{batch_outname}_provenance.tif"},
             },
             {
                 "name": "source_masks",
-                "args": {"res": increment, "output": f"{outname}_sources.vrt"},
+                "args": {"res": increment, "output": f"{batch_outname}_sources.vrt"},
             },
         ]
 
@@ -199,7 +221,7 @@ def build_cmd(
                     "mode": stack_mode,
                     "nodata": nodata,
                     "weight_threshold": "/".join([str(x) for x in weight_list]),
-                    "output": f"{outname}_stack.tif",
+                    "output": f"{batch_outname}_stack.tif",
                 },
             }
         )
@@ -281,7 +303,7 @@ def build_cmd(
             if "barrier" not in args:
                 args["barrier"] = "osm"
 
-        algo_hook.setdefault("args", {})["output"] = f"{outname}.tif"
+        algo_hook.setdefault("args", {})["output"] = f"{batch_outname}.tif"
         global_hooks.append(algo_hook)
 
         # --- Add Clipping (-C) ---
@@ -309,7 +331,7 @@ def build_cmd(
         global_hooks.append(
             {
                 "name": "viz_geoshade",
-                "args": {"output": f"{outname}_hs.tif", "cmap": "coastal_relief"},
+                "args": {"output": f"{batch_outname}_hs.tif", "cmap": "coastal_relief"},
             }
         )
 
@@ -340,14 +362,20 @@ def build_cmd(
                         "cells": ext_cells,
                         "pct": ext_pct,
                         "increment": increment,
-                        "outname": outname,
+                        "outname": batch_outname,
                     },
                 }
             ]
 
         # Ensure schemas validate the generated recipe
         config["schemas"] = [{"name": "validate-recipe"}]
 
+        if parsed_modifiers:
+            config["modifiers"].extend(parsed_modifiers)
+
+        if schema:
+            config["schemas"].extend(parsed_schemas)
+
         # --- Export or Execute ---
         if export:
             os.makedirs(base_outdir, exist_ok=True)
@@ -364,7 +392,12 @@ def build_cmd(
             recipe = Recipe.from_dict(config)
 
             # Fetchez handles all directory switching, batching, and execution
-            recipe.run(outdir=outdir, shared_cache=shared_cache, refresh=refresh)
+            recipe.run(
+                outdir=outdir,
+                shared_cache=shared_cache,
+                refresh=refresh,
+                ignore_failures=ignore_failures,
+            )
             click.secho(
                 "✨ Successfully completed Globato build pipeline!",
                 fg="green",

src/globato/hooks/filters/point_raster_mask.py

@@ -47,13 +47,17 @@ def setup(self, mod, entry):
             logger.warning(f"[{self.name}] No barrier provided. Skipping.")
             return False
 
+        region = getattr(mod, "region", None)
+        mod_outdir = getattr(mod, "outdir", getattr(mod, "_outdir", None))
+        cache_dir = mod_outdir if mod_outdir else os.getcwd()
+
         target_crs = entry.get("src_srs", "EPSG:4326")
         from globato.utils import resolve_barrier
 
         barrier_path = resolve_barrier(
             self.barrier,
-            region=getattr(mod, "region", None),
-            outdir=os.path.join(os.getcwd(), "auto_barriers"),
+            region=region,
+            outdir=os.path.join(cache_dir, "auto_barriers"),
             res=self.res,
             include_rivers=True,
             include_lakes=True,

src/globato/hooks/filters/rq.py

@@ -20,7 +20,7 @@
 from scipy.ndimage import map_coordinates
 
 import fetchez
-from fetchez.utils import str2inc
+from fetchez.utils import str2inc, parse_arg_to_list
 
 from .base import GlobatoFilter
 
@@ -82,8 +82,7 @@ def __init__(
         **kwargs,
     ):
         super().__init__(**kwargs)
-        self.ref_source = reference
-        self.ref_sources = [s.strip().lower() for s in str(reference).split("/")]
+        self.ref_sources = parse_arg_to_list(reference, str)
         self.threshold = float(threshold)
         self.mode = mode.lower()
         self.builder = builder.lower()
@@ -202,7 +201,7 @@ def _fetch_reference_files(self, region, outdir):
                 )
                 if files:
                     for f in files:
-                        if os.path.exists(f) and os.path.getsize(f) > 2000:
+                        if os.path.exists(f) and os.path.getsize(f) > 0:
                             valid_files.append(f)
             except Exception as e:
                 logger.warning(f"[RQ] Fetch failed for {source}: {e}")

src/globato/hooks/rasters/base.py

@@ -16,7 +16,6 @@
 import logging
 import shutil
 import numpy as np
-import tempfile
 import rasterio
 from rasterio.windows import Window
 import fiona
@@ -27,9 +26,6 @@
 logger = logging.getLogger(__name__)
 
 
-tmp_dir = tempfile.gettempdir()
-
-
 # =============================================================================
 # THE SHARED BASE (Utilities)
 # =============================================================================
@@ -144,14 +140,16 @@ def _get_barrier_geometries(
 
         mod = getattr(self, "current_mod", None)
         region = getattr(mod, "region", None) if mod else None
-        outdir = os.path.dirname(self.output) if self.output else os.getcwd()
+
+        mod_outdir = getattr(mod, "outdir", getattr(mod, "_outdir", None))
+        cache_dir = mod_outdir if mod_outdir else os.getcwd()
 
         from globato.utils import resolve_barrier
 
         barrier_path = resolve_barrier(
             self.barrier,
             region=region,
-            outdir=os.path.join(outdir, "auto_barriers"),
+            outdir=os.path.join(cache_dir, "auto_barriers"),
             output_type="vector",
             include_rivers=include_rivers,
             include_lakes=include_lakes,
@@ -313,11 +311,15 @@ def process_chunk(self, data, ndv, entry, transform=None, window=None):
 
     def run(self, entries):
         logger.info(f"[{self.name}] Running in local mode on {len(entries)} entries")
-
         new_entries = []
+
+        local_tmp = os.path.abspath("tmp")
+        os.makedirs(local_tmp, exist_ok=True)
+
         for mod, entry in entries:
             # SET CURRENT MOD FOR COASTLINE GENERATION
             self.current_mod = mod
+
             if self.has_stream(entry):
                 stream = entry.get("stream")
                 entry["stream"] = self._stream_wrapper(stream, entry)
@@ -337,7 +339,7 @@ def run(self, entries):
                 dst_fn = self.output
             else:
                 base_name = os.path.splitext(os.path.basename(src_fn))[0]
-                dst_fn = os.path.join(tmp_dir, f"{base_name}{self.suffix}.tif")
+                dst_fn = os.path.join(local_tmp, f"{base_name}{self.suffix}.tif")
 
             logger.debug(f"Running local {self.name} on {os.path.basename(src_fn)}")
             try:
@@ -414,6 +416,10 @@ def process_raster(self, src_path, dst_path, entry):
     def run(self, entries):
         logger.info(f"[{self.name}] Running in global mode on {len(entries)} entries")
         new_entries = []
+
+        local_tmp = os.path.abspath("tmp")
+        os.makedirs(local_tmp, exist_ok=True)
+
         for mod, entry in entries:
             # SET CURRENT MOD FOR COASTLINE GENERATION
             self.current_mod = mod
@@ -428,14 +434,12 @@ def run(self, entries):
 
                 base_name = os.path.basename(src_fn)
                 drain_fn = os.path.join(
-                    tmp_dir, f"{os.path.splitext(base_name)[0]}_drained_{self.name}.tif"
+                    local_tmp,
+                    f"{os.path.splitext(base_name)[0]}_drained_{self.name}.tif",
                 )
-                # drain_fn = f"{os.path.splitext(src_fn)[0]}_drained_{self.name}.tif"
                 entry["dst_fn"] = drain_fn
-
                 drainer = RasterWrite(suffix="", inline=False)
                 drainer.run([(mod, entry)])
-
                 src_fn = entry.get("dst_fn")
 
             if not src_fn or not os.path.exists(src_fn):
@@ -446,7 +450,7 @@ def run(self, entries):
                 dst_fn = self.output
             else:
                 base_name = os.path.splitext(os.path.basename(src_fn))[0]
-                dst_fn = os.path.join(tmp_dir, f"{base_name}{self.suffix}.tif")
+                dst_fn = os.path.join(local_tmp, f"{base_name}{self.suffix}.tif")
 
             logger.debug(f"Running global {self.name} on {os.path.basename(src_fn)}")
             try:

src/globato/hooks/rasters/binary_cudem.py

@@ -50,13 +50,12 @@ def __init__(
         resolutions=None,  # ["3s", "9s", "15s"],  # E.g., 1s=Dense, 3s=Med, 9s=Sparse
         algos=None,  # ,["raster_fill", "raster_fill", "interp_rbf"],
         blend_dists=None,  # 20,
-        barrier=None,
         decimation_mode="weighted_mean",
         bathy_max_z=-0.01,
-        keep_steps=True,
+        keep_steps=False,
         **kwargs,
     ):
-        super().__init__(barrier=barrier, strip_bands=True, **kwargs)
+        super().__init__(strip_bands=True, **kwargs)
 
         self.valid_algos = [
             "interp_gmt",
@@ -159,6 +158,9 @@ def _decimate_raster(self, src_path, dst_path, target_res):
 
         import fetchez
 
+        local_tmp = os.path.abspath("tmp")
+        os.makedirs(local_tmp, exist_ok=True)
+
         logger.info(
             f"[{self.name}] Decimating to {target_res} using '{self.decimation_mode}'..."
         )
@@ -171,6 +173,7 @@ def _decimate_raster(self, src_path, dst_path, target_res):
 
         decimated_stack = fetchez.get(
             "file",
+            outdir=local_tmp,
             region=region,
             region_srs=src_crs,
             path=src_path,