Analyzer

The InSARHub analyzer module provides workflow for InSAR time-series analysis.

• Import analyzer

  Import the Analyzer class to access all time-series analysis functionality

  from insarhub import Analyzer
  

• View Available Analyzers

  List all registered analyzers

  Analyzer.available()
  

Available Analyzers

Mintpy_SBAS_Base_AnalyzerHyp3_Mintpy_SBASISCE2_Mintpy_SBASGMTSAR_Mintpy_SBASGMTSAR_SBASISCE3_Dolphin_S1_PLISCE3_Dolphin_NISAR_PL

InSARHub wrapped Mintpy as one of its analysis backends. The Mintpy_SBAS_Base_Analyzer is implemented on top of a reusable base configuration class, which provides the full smallbaselineApp logic of Mintpy. Provides users with an experience similar to using MintPy directly, allowing full customization of processing parameters and steps.

Source code in src/insarhub/analyzer/mintpy_base.py

class Mintpy_SBAS_Base_Analyzer(BaseAnalyzer):

    description = "Generic MintPy SBAS analyzer, fully customizable configs."
    compatible_processor = 'all'
    default_config = Mintpy_SBAS_Base_Config
    '''
    Base class for Mintpy SBAS analysis. This class provides a template for implementing 
    specific analysis methods using the Mintpy software package.
    '''
    # Per-analyzer output folder under workdir (see MintPyPaths.subdir).
    # Subclasses override: "hyp3_mintpy" / "isce_mintpy" / "gmtsar_mintpy".
    MINTPY_SUBDIR = "mintpy"

    def __init__(self, config: Mintpy_SBAS_Base_Config | None = None):
        super().__init__(config)

        # absolute: MintPy's TimeSeriesAnalysis.open() os.chdir()s into
        # mintpy_dir, and its dask workers re-resolve paths in their own cwd --
        # a relative workdir then doubles up (".../gmtsar_mintpy/p100_f466/...")
        # and every stack path breaks. Found via a real run.
        self.workdir   = Path(self.config.workdir).expanduser().resolve()
        self._paths    = MintPyPaths(Path(self.workdir), type(self).MINTPY_SUBDIR)
        self._hyp3_paths = Hyp3Paths(Path(self.workdir))
        self.mintpy_dir = self._paths.mintpy_dir
        self.tmp_dir   = self._paths.tmp_dir
        self.clip_dir  = self._paths.clip_dir
        self.cfg_path  = self.mintpy_dir / '.mintpy.cfg'
        write_workflow_marker(self.workdir, analyzer=type(self).name)

    def prep_data(self):
        """Write the MintPy config file to workdir."""
        # not INSARHUB_CONTAINER_CHILD: when this already runs INSIDE the
        # container (re-invoked by _run_via_container), config.container is still
        # set, so without this guard it would `docker run` AGAIN inside the
        # MintPy image (which has no docker CLI) -- the nested call fails
        # silently, prep_data never resolves mintpy.load.*, and load_data then
        # dies with a missing ifgramStack.h5. The container side must run
        # prep_data locally. Mirrors the processor submit/retry guards.
        if self.config.container and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
            return self._run_via_container(["prep_data"])
        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        self._resolve_adaptive_coherence()
        self.config.write_mintpy_config(self.cfg_path)

    # ------------------------------------------------------------------ #
    #  Adaptive coherence thresholds                                      #
    # ------------------------------------------------------------------ #
    #: Value that asks InSARHub to derive a threshold from THIS stack.
    #: Distinct from "auto", which is MintPy's own token and resolves to its
    #: fixed defaults (0.7 network / 0.4 inversion) regardless of the data.
    ADAPTIVE = "adaptive"

    #: Ceilings on the adaptive coherence thresholds. The adaptive calc only
    #: needs to kick in for low-coherence stacks -- if the stack could support a
    #: threshold at or above the cap, that's already good and there's no reason
    #: to cut tighter, so it's clamped to the cap. In effect: only a value below
    #: the cap comes from the adaptive calculation; anything above uses the cap.
    ADAPTIVE_NETWORK_COH_CAP = 0.6     # network_minCoherence
    ADAPTIVE_MASK_COH_CAP = 0.6        # networkInversion_maskThreshold
    ADAPTIVE_REFERENCE_COH_CAP = 0.85  # reference_minCoherence

    def _pair_mean_coherence(self) -> dict[str, float]:
        """{pair_dir_name: mean coherence} read from the configured corFile
        glob. Works for any processor, since every SBAS analyzer sets
        mintpy.load.corFile to a per-pair glob."""
        import glob as _glob
        import numpy as np

        pattern = str(getattr(self.config, "load_corFile", "") or "")
        if not pattern or pattern == "auto":
            return {}
        out: dict[str, float] = {}
        for f in sorted(_glob.glob(pattern)):
            try:
                from osgeo import gdal
                gdal.UseExceptions()
                a = gdal.Open(f).ReadAsArray().astype("float32")
            except Exception:                                    # noqa: BLE001
                continue
            a = a[np.isfinite(a) & (a > 0)]
            if a.size:
                out[Path(f).parent.name] = float(a.mean())
        return out

    @staticmethod
    def _network_is_connected(pairs: list[tuple[str, str]], dates: set[str]) -> bool:
        """Every date reachable from any other via the kept interferograms."""
        if not dates:
            return False
        adj: dict[str, set[str]] = {d: set() for d in dates}
        for a, b in pairs:
            if a in adj and b in adj:
                adj[a].add(b); adj[b].add(a)
        seen, stack = set(), [next(iter(dates))]
        while stack:
            d = stack.pop()
            if d in seen:
                continue
            seen.add(d)
            stack.extend(adj[d] - seen)
        return seen == dates

    def _adaptive_min_coherence(self, coh: dict[str, float],
                                redundancy: float = 1.5) -> float | None:
        """The STRICTEST coherence threshold this stack can afford.

        A fixed threshold is a guess about absolute coherence, which varies
        hugely with band, land cover and season -- MintPy's 0.7 default is
        reasonable for a well-correlated C-band stack and discards everything
        here, where the best pair reaches 0.70 and the median is 0.47. With
        keepMinSpanTree on, that does not error; it silently collapses the
        network to its spanning tree, which removes every closed triplet and
        with them any redundancy for least squares or unwrapping-error repair.

        So rather than pick a number, pick the largest threshold that still
        leaves a network worth inverting:

          1. it must span every date (no isolated acquisition), and
          2. it must keep >= redundancy x (n_dates - 1) interferograms --
             1.5 means half again as many as a spanning tree, so closed
             triplets survive.

        Returns None when the stack cannot satisfy both, leaving the
        configured value untouched rather than inventing one.
        """
        if len(coh) < 2:
            return None
        pairs = {}
        for name, c in coh.items():
            parts = name.replace("-", "_").split("_")
            if len(parts) == 2:
                pairs[name] = (parts[0], parts[1])
        if len(pairs) < 2:
            return None
        dates = {d for p in pairs.values() for d in p}
        need = max(1, int(round(redundancy * (len(dates) - 1))))

        best = None
        for t in sorted(set(round(c, 3) for c in coh.values())):
            kept = [pairs[n] for n, c in coh.items() if c >= t and n in pairs]
            if len(kept) >= need and self._network_is_connected(kept, dates):
                best = t
        if best is None:
            return None
        # Cap at ADAPTIVE_NETWORK_COH_CAP: the adaptive sweep only matters for
        # low-coherence stacks. If it found the network survives a threshold at
        # or above the cap, that's already a good network -- clamp to the cap
        # rather than cut tighter. Lowering the threshold only keeps MORE pairs,
        # so the clamped network is still connected + redundant.
        return min(best, self.ADAPTIVE_NETWORK_COH_CAP)

    def _pixel_mean_coherence(self):
        """Per-pixel mean spatial coherence across the stack (2-D array), read
        from the configured corFile glob -- MintPy's avgSpatialCoherence before
        it exists. Returns None if the grids are missing or not co-registered."""
        import glob as _glob
        import numpy as np

        pattern = str(getattr(self.config, "load_corFile", "") or "")
        if not pattern or pattern == "auto":
            return None
        stack = []
        for f in sorted(_glob.glob(pattern)):
            try:
                from osgeo import gdal
                gdal.UseExceptions()
                a = gdal.Open(f).ReadAsArray().astype("float32")
            except Exception:                                    # noqa: BLE001
                continue
            a[~np.isfinite(a)] = np.nan
            a[a <= 0] = np.nan
            stack.append(a)
        if not stack or len({s.shape for s in stack}) != 1:
            return None
        return np.nanmean(np.stack(stack), axis=0)

    def _adaptive_mask_threshold(self, keep_frac: float = 0.85) -> float | None:
        """A networkInversion.maskThreshold that keeps ~keep_frac of the stack's
        coherence OBSERVATIONS in the inversion.

        maskThreshold masks each pixel out of each interferogram whose spatial
        coherence is below it. MintPy's fixed 0.4 (and any value at or above the
        stack's own coherence) drops nearly every observation on a low-coherence
        GMTSAR stack -> numInvIfgram 0 -> an all-zero time series and velocity.
        Set it at the (1-keep_frac) percentile of the POOLED per-pair coherence
        so most observations survive; None if no corFile grids are readable."""
        import glob as _glob
        import numpy as np

        pattern = str(getattr(self.config, "load_corFile", "") or "")
        if not pattern or pattern == "auto":
            return None
        vals = []
        for f in sorted(_glob.glob(pattern)):
            try:
                from osgeo import gdal
                gdal.UseExceptions()
                a = gdal.Open(f).ReadAsArray().astype("float32")
            except Exception:                                    # noqa: BLE001
                continue
            a = a[np.isfinite(a) & (a > 0)]
            if a.size:
                vals.append(a)
        if not vals:
            return None
        pooled = np.concatenate(vals)
        thr = float(np.percentile(pooled, (1.0 - keep_frac) * 100.0))
        # Cap like network_minCoherence: only a value below the cap comes from
        # the adaptive percentile; at/above it, use the cap.
        return round(min(self.ADAPTIVE_MASK_COH_CAP, max(0.0, thr)), 3)

    def _adaptive_reference_min_coherence(self) -> float | None:
        """A reference.minCoherence floor derived from THIS stack.

        The reference point is chosen among pixels whose average spatial
        coherence exceeds this floor; MintPy's fixed 0.85 finds no pixel at all
        on a low-coherence stack and errors. Instead pick the level that keeps
        the most-coherent ~2% of pixels as reference candidates, clamped so it
        never demands more than MintPy's 0.85 nor drops below 0.30 (a reference
        pixel should still be genuinely reliable)."""
        import numpy as np
        m = self._pixel_mean_coherence()
        if m is None:
            return None
        v = m[np.isfinite(m)]
        if v.size < 100:
            return None
        thr = float(np.nanpercentile(v, 98))
        # below the cap use the adaptive percentile; at/above it use the cap.
        return round(min(self.ADAPTIVE_REFERENCE_COH_CAP, max(0.30, thr)), 2)

    def _resolve_adaptive_coherence(self) -> None:
        """Replace ADAPTIVE sentinels with values derived from the stack."""
        pair_fields = [f for f in ("network_minCoherence",
                                   "networkInversion_maskThreshold")
                       if str(getattr(self.config, f, "")).lower() == self.ADAPTIVE]
        ref_adaptive = (str(getattr(self.config, "reference_minCoherence", "")).lower()
                        == self.ADAPTIVE)
        if not pair_fields and not ref_adaptive:
            return

        import numpy as np
        # Per-pair thresholds: network selection + inversion pixel mask.
        if pair_fields:
            coh = self._pair_mean_coherence()
            if not coh:
                logger.warning("adaptive coherence: no per-pair coherence found via "
                               "mintpy.load.corFile; leaving thresholds unchanged")
                for f in pair_fields:
                    setattr(self.config, f, "auto")
            else:
                vals = np.array(list(coh.values()))
                thr = self._adaptive_min_coherence(coh)
                print(f"{Fore.CYAN}Adaptive coherence: {len(coh)} pairs, "
                      f"coherence {vals.min():.2f}–{vals.max():.2f} "
                      f"(median {np.median(vals):.2f}){Fore.RESET}")
                if thr is None:
                    logger.warning("adaptive coherence: no threshold keeps the network "
                                   "connected with redundancy; falling back to MintPy auto")
                    for f in pair_fields:
                        setattr(self.config, f, "auto")
                else:
                    for f in pair_fields:
                        if f == "network_minCoherence":
                            setattr(self.config, f, round(float(thr), 2))
                            kept = int((vals >= thr).sum())
                            print(f"  network.minCoherence -> {thr:.2f}  "
                                  f"(keeps {kept}/{len(coh)} interferograms)")
                        else:
                            # Pixel mask threshold: this masks per-pixel-per-
                            # ifgram OBSERVATIONS out of the inversion, so it
                            # must sit BELOW the stack's own coherence -- a fixed
                            # 0.4/0.5 (or even a 0.1 floor) on a low-coherence
                            # GMTSAR stack drops nearly every observation, leaving
                            # numInvIfgram=0 and an all-zero time series. Derive
                            # it from the observation distribution so ~85% survive.
                            px = self._adaptive_mask_threshold()
                            if px is None:                       # no corr grids
                                px = round(max(0.0, float(thr) * 0.5), 3)
                            setattr(self.config, f, px)
                            print(f"  networkInversion.maskThreshold -> {px:.3f} "
                                  f"(keeps ~85% of coherence observations)")

        # Reference-point floor: per-PIXEL average coherence, not per-pair.
        if ref_adaptive:
            rthr = self._adaptive_reference_min_coherence()
            if rthr is None:
                logger.warning("adaptive reference.minCoherence: could not derive "
                               "from per-pixel coherence; falling back to MintPy auto")
                setattr(self.config, "reference_minCoherence", "auto")
            else:
                setattr(self.config, "reference_minCoherence", rthr)
                print(f"  reference.minCoherence -> {rthr:.2f}")

    def _cfg_load_paths_resolved(self) -> bool:
        """True once prep_data has written a real ``mintpy.load.unwFile`` into
        the cfg (i.e. not still ``auto``). Used to decide whether load_data can
        run on its own or needs prep_data first."""
        if not self.cfg_path.exists():
            return False
        for ln in self.cfg_path.read_text().splitlines():
            if ln.strip().startswith("mintpy.load.unwFile"):
                v = ln.partition("=")[2].strip()
                return v not in ("auto", "", "None")
        return False

    def _sync_runtime_cfg(self) -> None:
        """Rewrite ``.mintpy.cfg`` from the current (possibly overridden) config,
        preserving whatever ``prep_data`` computed only into the file.

        ``prep_data`` writes the geocoded load paths (``mintpy.load.unwFile`` …),
        the resolved ``metaFile``/``demFile``/``baselineDir`` and an appended
        ``HEADING`` straight into the cfg. A per-step process (e.g. a container
        ``--step invert_network``) has all of those back at ``"auto"`` on
        ``self.config``, so a blind ``write_mintpy_config`` would clobber them.
        Instead: emit a fresh cfg from config, but for any ``mintpy.load.*`` key
        the config leaves unset keep the file's value, and carry over any key the
        config never emits at all (``HEADING``). Everything else — the network /
        inversion / correction knobs a user overrode — is taken from config.
        """
        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        tmp = self.mintpy_dir / ".mintpy.cfg.tmp"
        self.config.write_mintpy_config(tmp)
        new_lines = tmp.read_text().splitlines()
        tmp.unlink()

        def _parse(text: str) -> dict[str, str]:
            d: dict[str, str] = {}
            for ln in text.splitlines():
                if "=" in ln and not ln.strip().startswith("#"):
                    k, _, v = ln.partition("=")
                    d[k.strip()] = v.strip()
            return d

        existing = _parse(self.cfg_path.read_text()) if self.cfg_path.exists() else {}

        out, seen = [], set()
        for ln in new_lines:
            if "=" in ln and not ln.strip().startswith("#"):
                k = ln.partition("=")[0].strip()
                v = ln.partition("=")[2].strip()
                seen.add(k)
                # Keep the value prep_data resolved into the file when this
                # per-step process only has a placeholder on self.config: the
                # geocoded load paths (still "auto" here) and any coherence
                # threshold prep_data derived from the stack (still "adaptive").
                if k in existing and (
                        (k.startswith("mintpy.load.") and v in ("auto", "", "None"))
                        or v.lower() == self.ADAPTIVE):
                    out.append(f"{k:<40} = {existing[k]}")
                    continue
            out.append(ln)
        # keep cfg-only keys the config never emits (e.g. HEADING)
        for k, v in existing.items():
            if k not in seen:
                out.append(f"{k:<40} = {v}")
        self.cfg_path.write_text("\n".join(out) + "\n")

    def _validate_cds_token(self, key: str) -> bool:
        """Validate a CDS API token via a lightweight HTTP request (no download)."""
        import requests as _requests
        endpoints = [
            # Fast profile endpoint (new CDS API)
            ("GET", "https://cds.climate.copernicus.eu/api/account/me",
             {"PRIVATE-TOKEN": key}),
            # Fallback: jobs list
            ("GET", "https://cds.climate.copernicus.eu/api/retrieve/v1/jobs",
             {"PRIVATE-TOKEN": key}),
        ]
        for method, url, headers in endpoints:
            try:
                resp = _requests.request(method, url, headers=headers,
                                         params={"limit": 1}, timeout=30)
                if resp.status_code == 200:
                    return True
                if resp.status_code in (401, 403):
                    return False
            except _requests.exceptions.Timeout:
                continue
            except Exception:
                continue
        # If all endpoints timed out, assume valid to avoid blocking the user
        print(f"{Fore.YELLOW}CDS API unreachable (timeout) — assuming token is valid.{Fore.RESET}")
        return True

    def _cds_authorize(self):
        """Ensure valid CDS credentials exist, prompting the user if needed."""
        cdsapirc_path = Path.home() / ".cdsapirc"
        # Try existing .cdsapirc first
        if cdsapirc_path.is_file():
            key = None
            for line in cdsapirc_path.read_text().splitlines():
                if line.strip().startswith("key:"):
                    key = line.split(":", 1)[1].strip()
                    break
            if key and self._validate_cds_token(key):
                return True
            print(f"{Fore.YELLOW}CDS token in .cdsapirc is invalid or expired. Will prompt login.\n")

        # Prompt user for a valid token
        while True:
            self._cds_token = getpass.getpass("Enter your CDS api token at https://cds.climate.copernicus.eu/profile: ")
            if not self._validate_cds_token(self._cds_token):
                print(f"{Fore.RED}Authentication failed. Please check your token and try again.\n")
                continue
            cdsapirc_path.write_text(f"url: https://cds.climate.copernicus.eu/api\nkey: {self._cds_token}\n")
            print(f"{Fore.GREEN}Credentials saved to {cdsapirc_path}.\n")
            return True

    def _serialize_config_overrides(self) -> str:
        """Serialize non-default config fields back to '--flag value' CLI args.

        Used to re-invoke `insarhub analyzer ... run` (via SLURM or a
        container) with the same resolved config. `container` itself is
        always excluded — it's a per-invocation flag, and including it here
        would make a container/HPC re-invocation try to launch another
        nested container.
        """
        _skip = {"name", "workdir", "debug", "hpc_mode", "container"}
        config_cls = type(self.config)
        defaults = {}
        for f in dataclasses.fields(config_cls):
            if f.default is not dataclasses.MISSING:
                defaults[f.name] = f.default
            elif f.default_factory is not dataclasses.MISSING:
                defaults[f.name] = f.default_factory()

        override_flags = []
        for f in dataclasses.fields(config_cls):
            if f.name in _skip:
                continue
            val = getattr(self.config, f.name)
            if val == defaults.get(f.name):
                continue
            if isinstance(val, bool):
                if val:
                    override_flags.append(f"--{f.name}")
            elif isinstance(val, (list, tuple)):
                override_flags.append(f"--{f.name} " + " ".join(str(v) for v in val))
            elif isinstance(val, dict):
                override_flags.append(f"--{f.name} '{json.dumps(val)}'")
            elif val is not None:
                override_flags.append(f"--{f.name} {val}")

        return (" " + " ".join(override_flags)) if override_flags else ""

    @staticmethod
    def _plot_result_safe(app, max_attempts: int = 3) -> None:
        """Call app.plot_result(), retrying past a known matplotlib race.

        MintPy's plot_result() parallelizes per-file plotting via
        joblib.Parallel over view.py calls that all use pyplot's global,
        not-thread-safe figure registry -- two calls landing close enough
        together can both try to claim the same figure number. Pre-3.11
        matplotlib silently warned and reused the figure; matplotlib >=3.11
        made that a hard ValueError ("Figure N already exists..."),
        turning a harmless race into a crash. Since the underlying SBAS
        numbers (velocity.h5, timeseries.h5, etc.) are already fully
        computed by this point -- only the pic/ figures are at risk -- a
        plain retry is cheap and usually succeeds (view.py's own --update
        flag skips replotting files already written by the failed attempt).
        """
        import matplotlib

        for attempt in range(1, max_attempts + 1):
            try:
                app.plot_result()
                return
            except ValueError as e:
                msg = str(e)
                if "already exists" not in msg or "igure" not in msg:
                    raise  # not the known matplotlib figure-registry race
                if attempt < max_attempts:
                    print(
                        f"{Fore.YELLOW}[WARNING] Plotting hit a known matplotlib "
                        f"{matplotlib.__version__} bug (parallel figure-number "
                        f"race, see plt.figure()'s strict num-reuse check added "
                        f"in matplotlib 3.11) -- retrying ({attempt}/{max_attempts - 1})...{Fore.RESET}"
                    )
                else:
                    print(
                        f"{Fore.YELLOW}[WARNING] SBAS analysis succeeded, but "
                        f"plotting failed after {max_attempts} attempts due to a "
                        f"known matplotlib {matplotlib.__version__} bug ('{msg}'). "
                        f"All numerical results (velocity.h5, timeseries.h5, etc.) "
                        f"are complete and valid -- only mintpy_dir/pic/ figures "
                        f"may be missing/incomplete. Re-run with '--step plot' to "
                        f"try generating them again, or install matplotlib<3.11 "
                        f"to eliminate the underlying race entirely.{Fore.RESET}"
                    )

    def _run_via_container(self, steps: list[str] | None = None) -> None:
        """Re-invoke `insarhub analyzer ... run` inside self.config.container.

        The container image is expected to have `insarhub` (plus MintPy)
        installed — mirrors ISCE2_Base._reinvoke_via_container's approach for
        the processor side.
        """
        from insarhub.utils.container import wrap_container_cmd

        step_args = f" --step {' '.join(steps)}" if steps else ""
        extra = self._serialize_config_overrides()
        cli_cmd = f"insarhub analyzer -N {type(self).name} -w {self.workdir} run{step_args}{extra}"
        wrapped = wrap_container_cmd(self.config.container, cli_cmd, Path(self.workdir))

        result = subprocess.run(wrapped, shell=True)
        if result.returncode != 0:
            raise RuntimeError(f"Container run failed (exit {result.returncode}): {wrapped}")

    def submit_hpc(self, steps: list[str] | None = None) -> str | None:
        """Generate a sbatch script for the full MintPy run and submit it.

        Returns the SLURM job ID string, or None if sbatch_options.json was
        just created/updated and needs review before submitting — callers
        must check for this and stop rather than treat it as success.
        """
        from insarhub.utils.tool import Slurmjob_Config
        from insarhub.processor.isce2_base import (
            _merge_sbatch_opts, _SBATCH_DEFAULT_TEMPLATE, load_or_init_sbatch_options,
        )

        mintpy_dir = self._paths.mintpy_dir
        mintpy_dir.mkdir(parents=True, exist_ok=True)

        # This method is shared by ISCE2_Mintpy_SBAS/Hyp3_Mintpy_SBAS/GMTSAR_Mintpy_SBAS, but
        # both the sbatch_options.json step key and a *fresh* file's initial
        # content follow whichever processor the workdir actually uses (each
        # subclass's compatible_processor says which). ISCE's own step keys
        # are stackSentinel's run-file numbers, where SBAS is the 17th and
        # last step -- a GMTSAR workdir has no such numbering (its stages are
        # named align/topo/intf/merge), so "17" there would be a meaningless
        # borrowed label; it uses "sbas" instead.
        if self.compatible_processor == "GMTSAR_S1":
            from insarhub.processor.gmtsar_s1 import _GMTSAR_SBATCH_DEFAULT_TEMPLATE
            default_template = _GMTSAR_SBATCH_DEFAULT_TEMPLATE
            step_key = "sbas"
        else:
            default_template = _SBATCH_DEFAULT_TEMPLATE
            step_key = "17"
        per_step = load_or_init_sbatch_options(
            Path(self.workdir), step_key, "SBAS", default_template=default_template)
        if per_step is None:
            return None
        opts = _merge_sbatch_opts(per_step, step_key)

        _slurm_fields = {f.name for f in dataclasses.fields(Slurmjob_Config)}
        _skip = {"job_name", "output_file", "error_file", "command",
                 "modules", "conda_env", "export_env", "array", "dependency"}
        slurm_kwargs = {k: v for k, v in opts.items()
                        if k in _slurm_fields and k not in _skip}

        slurm_cfg = Slurmjob_Config(
            job_name="mintpy_sbas",
            output_file=str(mintpy_dir / "mintpy_slurm_%j.out"),
            error_file=str(mintpy_dir / "mintpy_slurm_%j.err"),
            **slurm_kwargs,
        )

        import os
        import shutil

        insarhub_bin = shutil.which("insarhub") or f"{Path(sys.executable).parent}/insarhub"
        analyzer_name = type(self).name
        current_path  = os.environ.get("PATH", "")

        step_args = ""
        if steps:
            step_args = " --step " + " ".join(steps)

        # Serialize non-default config overrides back to CLI flags so that
        # prep_data inside SLURM writes the correct .mintpy.cfg values.
        extra = self._serialize_config_overrides()

        body_cmd = f"{insarhub_bin} analyzer -N {analyzer_name} -w {self.workdir} run{step_args}{extra}"
        if self.config.container:
            from insarhub.utils.container import wrap_container_cmd
            body_cmd = wrap_container_cmd(self.config.container, body_cmd, Path(self.workdir))

        body = "\n".join([
            f'export PATH="{current_path}"',
            body_cmd,
        ])

        lines = ["#!/bin/bash"] + slurm_cfg.to_header_lines() + ["", body, ""]
        sbatch_script = mintpy_dir / "mintpy_sbas.sbatch"
        sbatch_script.write_text("\n".join(lines) + "\n")
        sbatch_script.chmod(0o755)

        result = subprocess.run(
            ["sbatch", "--parsable", str(sbatch_script)],
            capture_output=True, text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(f"sbatch failed: {result.stderr.strip()}")

        job_id = result.stdout.strip().split(";")[0]

        job_file = mintpy_dir / "mintpy_job.json"
        job_file.write_text(json.dumps({
            "job_id":  job_id,
            "status":  "PENDING",
            "script":  str(sbatch_script),
            "log":     str(mintpy_dir / f"mintpy_slurm_{job_id}.out"),
        }, indent=2))

        print(f"{Fore.GREEN}MintPy SBAS job submitted: {job_id}{Style.RESET_ALL}")
        print(f"  script : {sbatch_script}")
        print(f"  log    : {mintpy_dir}/mintpy_slurm_{job_id}.out")

        return job_id
        return job_id

    def run(self, steps=None):
        """
        Run the MintPy SBAS time-series analysis workflow.

        This method writes the MintPy configuration file, optionally authorizes
        CDS access for tropospheric correction, and executes the selected
        MintPy processing steps using TimeSeriesAnalysis.

        Args:
            steps (list[str] | None, optional):
                List of MintPy processing steps to execute. If None, the
                default full workflow is executed:
                    [
                        'load_data', 'modify_network', 'reference_point', 'quick_overview',
                        'correct_unwrap_error', 'invert_network', 'correct_LOD', 'correct_SET',
                        'correct_ionosphere', 'correct_troposphere',
                        'deramp', 'correct_topography', 'residual_RMS',
                        'reference_date', 'velocity', 'geocode',
                        'google_earth', 'hdfeos5'
                    ]

        Raises:
            RuntimeError: If tropospheric delay method requires CDS authorization
                and authorization fails.
            Exception: Propagates exceptions raised during MintPy execution.

        Notes:
            - If `troposphericDelay_method` is set to 'pyaps', CDS
            authorization is performed before running MintPy.
            - The configuration file is written to `self.cfg_path`.
            - Processing is executed inside `self.workdir`.
            - This method wraps MintPy TimeSeriesAnalysis for SBAS workflows.
        """
        # HPC: hand the whole analysis to SLURM instead of running MintPy in this
        # process -- mirrors processor.submit()'s hpc dispatch so the API is
        # symmetric (set hpc_mode, call run()). Returns submit_hpc()'s job id, or
        # None if it just wrote sbatch_options.json for review (call run() again
        # after tuning it). The sbatch body re-invokes `insarhub analyzer ... run`
        # WITHOUT --hpc-mode (hpc_mode is skipped by _serialize_config_overrides),
        # so the compute-node run() sees hpc_mode=False and runs locally -- no
        # resubmission loop. Guarded off inside a container child for the same
        # reason (hpc_mode isn't carried in there either).
        if getattr(self.config, "hpc_mode", False) and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
            return self.submit_hpc(steps=steps)

        # not INSARHUB_CONTAINER_CHILD: run the steps locally when already inside
        # the container (see prep_data's guard for the full rationale).
        if self.config.container and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
            return self._run_via_container(steps)

        run_steps = steps or [
            'load_data', 'modify_network', 'reference_point', 'quick_overview',
            'correct_unwrap_error', 'invert_network',
            'correct_LOD', 'correct_SET', 'correct_ionosphere', 'correct_troposphere',
            'deramp', 'correct_topography', 'residual_RMS', 'reference_date',
            'velocity', 'geocode', 'google_earth', 'hdfeos5'
        ]

        # prep_data is what fills mintpy.load.* with the real geocoded file
        # paths (plus the resolved adaptive thresholds and HEADING). The GUI
        # lets users deselect it, and load_data can be run on its own, so
        # self-heal: if it isn't in this run and the cfg still has no resolved
        # load paths, run prep_data first. Otherwise MintPy finds no files,
        # writes no ifgramStack.h5, and load_data fails. prep_data is cheap to
        # repeat (cached DEM / baselines).
        if 'prep_data' not in run_steps and not self._cfg_load_paths_resolved():
            print(f"{Fore.YELLOW}mintpy.load.* not resolved yet — running prep_data "
                  f"first to set the file locations.{Fore.RESET}")
            self.prep_data()

        if not self.cfg_path.exists():
            print(f"{Fore.YELLOW}Warning: .mintpy.cfg not found — writing config now. "
                  f"If this is a Hyp3_Mintpy_SBAS run, make sure 'prep_data' (or '--step prep') "
                  f"was completed first so load parameters are correct.{Fore.RESET}")
        # Re-apply the (possibly CLI-/GUI-overridden) config to .mintpy.cfg on
        # every run, not just the first: prep_data creates the file, so without
        # this any parameter passed to a later step (e.g. --networkInversion_
        # minTempCoh on invert_network) was silently dropped because the stale
        # file already existed. Preserves the load paths / HEADING prep_data
        # computed into the file (they are not on self.config here).
        self._sync_runtime_cfg()

        if self.config.troposphericDelay_method == 'pyaps' and 'correct_troposphere' in run_steps:
            self._cds_authorize()
        print(f'{Style.BRIGHT}{Fore.MAGENTA}Running MintPy Analysis...{Fore.RESET}')
        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        _patch_mintpy_plot_bugs()
        from mintpy.smallbaselineApp import TimeSeriesAnalysis
        app = TimeSeriesAnalysis(self.cfg_path.as_posix(), self.mintpy_dir.as_posix())
        try:
            app.open()
            app.run(steps=run_steps)
            if 'geocode' in run_steps:
                self._geocode_diagnostic_files(self.mintpy_dir)
            # Mirrors mintpy.smallbaselineApp's own CLI wrapper
            # (run_smallbaselineApp()), which calls these two after run() --
            # plot_result() is what actually populates mintpy_dir/pic/, and
            # close() is what restores the process's working directory after
            # open() changed into mintpy_dir (skipping it would leave a
            # long-running server process permanently cd'd into the last
            # analyzed folder).
            if app.template.get('mintpy.plot') and len(run_steps) > 1:
                self._plot_result_safe(app)
        finally:
            app.close()

    def plot(self) -> None:
        """(Re)generate the figures under mintpy_dir/pic/ from already-computed results.

        run()'s own post-run plotting only fires for a single bulk multi-step
        call (mirroring MintPy's own CLI semantics: len(run_steps) > 1). Both
        the CLI (`analyzer run`) and the GUI execute steps one at a time
        internally for per-step progress reporting, so that condition never
        actually triggers there — this method is the explicit, standalone
        alternative both call once after their step sequence completes
        (or on-demand, e.g. the GUI's "plot" checkbox / CLI's `--step plot`).
        """
        if self.config.container and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
            return self._run_via_container(['plot'])
        if not self.cfg_path.exists():
            raise FileNotFoundError(
                f"{self.cfg_path} not found — run prep_data and at least "
                f"load_data/invert_network/velocity before plotting."
            )
        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        _patch_mintpy_plot_bugs()
        from mintpy.smallbaselineApp import TimeSeriesAnalysis
        app = TimeSeriesAnalysis(self.cfg_path.as_posix(), self.mintpy_dir.as_posix())
        try:
            app.open()
            self._plot_result_safe(app)
        finally:
            app.close()

    def _geocode_diagnostic_files(self, mintpy_work: Path) -> None:
        """Geocode diagnostic files omitted from MintPy's default geocode step.

        MintPy only geocodes temporalCoherence, avgSpatialCoh, timeseries, velocity.
        avgPhaseVelocity, numTriNonzeroIntAmbiguity, and maskConnComp are left in
        radar coordinates. This method geocodes them into geo/ when a lookup table
        is available (radar-coord inputs). For already-geocoded inputs the method
        is a no-op.
        """
        geo_dir = mintpy_work / 'geo'
        if not geo_dir.exists():
            return  # geocode step skipped by MintPy (inputs already geocoded)

        try:
            from mintpy.utils import utils as _mut
            _, _, lookup_file = _mut.check_loaded_dataset(str(mintpy_work), print_msg=False)[:3]
        except Exception:
            return

        if not lookup_file:
            return  # geocoded inputs — no lookup table

        _DIAG = ['avgPhaseVelocity.h5', 'numTriNonzeroIntAmbiguity.h5', 'maskConnComp.h5']
        to_geo = [
            str(mintpy_work / f) for f in _DIAG
            if (mintpy_work / f).exists() and not (geo_dir / f'geo_{f}').exists()
        ]
        if not to_geo:
            return

        try:
            import mintpy.cli.geocode as _geo_cli
            iargs = to_geo + ['-l', lookup_file, '--outdir', str(geo_dir), '--update']
            print(f'{Fore.CYAN}Geocoding diagnostic files: {[Path(f).name for f in to_geo]}{Fore.RESET}')
            _geo_cli.main(iargs)
        except Exception as e:
            print(f'{Fore.YELLOW}Warning: could not geocode diagnostic files: {e}{Fore.RESET}')

    def cleanup(self):
        """
        Remove temporary files and directories generated during processing.

        This method deletes the temporary working directories and any `.zip`
        archives in `self.workdir`. If debug mode is enabled, temporary files
        are preserved and a message is printed instead.

        Behavior:
            - Deletes `self.tmp_dir` and `self.clip_dir` if they exist.
            - Deletes all `.zip` files in `self.workdir`.
            - Prints informative messages for each removal or failure.
            - Respects `self.config.debug`; no files are deleted in debug mode.

        Raises:
            Exception: Propagates any unexpected errors raised during removal.

        Notes:
            - Useful for freeing disk space after large InSAR or MintPy
            processing workflows.
            - Temporary directories should contain only non-essential files
            to avoid accidental data loss.
        """

        if self.config.debug:
            print(f"{Fore.YELLOW}Debug mode is enabled. Keeping temporary files at: {self.workdir}{Fore.RESET}")
            return
        print(f"{Fore.CYAN}Step: Cleaning up temporary directories...{Fore.RESET}")

        for folder in [self.tmp_dir, self.clip_dir]:
            if folder.exists() and folder.is_dir():
                try:
                    shutil.rmtree(folder)
                    print(f"  Removed: {folder.relative_to(self.workdir)}")
                except Exception as e:
                    print(f"{Fore.RED}  Failed to remove {folder}: {e}{Fore.RESET}")

        _hyp3_dir = self._hyp3_paths.output_dir
        zips = list(_hyp3_dir.glob('*.zip')) if _hyp3_dir.exists() else list(Path(self.workdir).glob('*.zip'))
        if zips:
            print(f"{Fore.CYAN}Step: Removing zip archives...{Fore.RESET}")
            for zf in zips:
                try:
                    zf.unlink()
                    print(f"  Removed: {zf.name}")
                except Exception as e:
                    print(f"{Fore.RED}  Failed to remove {zf.name}: {e}{Fore.RESET}")

        print(f"{Fore.GREEN}Cleanup complete.{Fore.RESET}")

Usage

• Create Analyzer with Parameters

  Initialize an analyzer instance

  analyzer = Analyzer.create('Mintpy_SBAS_Base_Analyzer',
                              workdir="/your/work/dir",
                              load_processor="hyp3", ....)
  

  OR

  params = {"workdir": "/your/work/dir", "load_processor": "hyp3" ....}
  analyzer = Analyzer.create('Mintpy_SBAS_Base_Analyzer', **params)
  

  OR

  from insarhub.config import Mintpy_SBAS_Base_Config
  cfg = Mintpy_SBAS_Base_Config(workdir="/your/work/dir",
                                load_processor="hyp3",
                                ....)
  analyzer = Analyzer.create('Mintpy_SBAS_Base_Analyzer', config=cfg)
  

  The base config Mintpy_SBAS_Base_Config contains all parameters from Mintpy smallbaselineApp.cfg. For detailed descriptions refer to the official Mintpy config documentation.

  Source code in src/insarhub/config/defaultconfig.py

  @dataclass
  class Mintpy_SBAS_Base_Config:
      '''
      Dataclass containing all configuration options for Mintpy SBAS jobs.
  
      UI metadata is stored in ``_ui_groups`` / ``_ui_fields`` and consumed
      by the API layer to auto-generate the settings panel.
      '''
  
      # ── UI metadata consumed by the API / settings panel ─────────────────────
      _ui_groups: ClassVar[list] = [
          {"label": "Compute Resources",
           "fields": ["compute_maxMemory", "compute_cluster", "compute_numWorker", "compute_config"]},
          {"label": "Load Data",
           "fields": ["load_processor", "load_autoPath", "load_updateMode", "load_compression",
                      "load_metaFile", "load_baselineDir",
                      "load_unwFile", "load_corFile", "load_connCompFile", "load_intFile", "load_magFile",
                      "load_ionUnwFile", "load_ionCorFile", "load_ionConnCompFile",
                      "load_azOffFile", "load_rgOffFile", "load_azOffStdFile", "load_rgOffStdFile", "load_offSnrFile",
                      "load_demFile", "load_lookupYFile", "load_lookupXFile",
                      "load_incAngleFile", "load_azAngleFile", "load_shadowMaskFile", "load_waterMaskFile", "load_bperpFile",
                      "subset_yx", "subset_lalo",
                      "multilook_method", "multilook_ystep", "multilook_xstep"]},
          {"label": "Modify Network",
           "fields": ["network_tempBaseMax", "network_perpBaseMax", "network_connNumMax",
                      "network_startDate", "network_endDate", "network_excludeDate", "network_excludeDate12",
                      "network_excludeIfgIndex", "network_referenceFile",
                      "network_coherenceBased", "network_minCoherence",
                      "network_areaRatioBased", "network_minAreaRatio",
                      "network_keepMinSpanTree", "network_maskFile", "network_aoiYX", "network_aoiLALO"]},
          {"label": "Reference Point",
           "fields": ["reference_yx", "reference_lalo", "reference_maskFile",
                      "reference_coherenceFile", "reference_minCoherence"]},
          {"label": "Unwrap Error Correction",
           "fields": ["unwrapError_method", "unwrapError_waterMaskFile", "unwrapError_connCompMinArea",
                      "unwrapError_numSample", "unwrapError_ramp", "unwrapError_bridgePtsRadius"]},
          {"label": "Network Inversion",
           "fields": ["networkInversion_weightFunc", "networkInversion_waterMaskFile",
                      "networkInversion_minNormVelocity", "networkInversion_maskDataset",
                      "networkInversion_maskThreshold", "networkInversion_minRedundancy",
                      "networkInversion_minTempCoh", "networkInversion_minNumPixel", "networkInversion_shadowMask"]},
          {"label": "Solid Earth Tides",
           "fields": ["solidEarthTides"]},
          {"label": "Ionosphere Correction",
           "fields": ["ionosphericDelay_method", "ionosphericDelay_excludeDate", "ionosphericDelay_excludeDate12"]},
          {"label": "Troposphere Correction",
           "fields": ["troposphericDelay_method", "troposphericDelay_weatherModel", "troposphericDelay_weatherDir",
                      "troposphericDelay_polyOrder", "troposphericDelay_looks", "troposphericDelay_minCorrelation",
                      "troposphericDelay_gacosDir"]},
          {"label": "Deramp",
           "fields": ["deramp", "deramp_maskFile"]},
          {"label": "Topography Correction",
           "fields": ["topographicResidual", "topographicResidual_polyOrder", "topographicResidual_phaseVelocity",
                      "topographicResidual_stepDate", "topographicResidual_excludeDate",
                      "topographicResidual_pixelwiseGeometry"]},
          {"label": "Residual RMS",
           "fields": ["residualRMS_maskFile", "residualRMS_deramp", "residualRMS_cutoff"]},
          {"label": "Reference Date",
           "fields": ["reference_date"]},
          {"label": "Velocity",
           "fields": ["timeFunc_startDate", "timeFunc_endDate", "timeFunc_excludeDate",
                      "timeFunc_polynomial", "timeFunc_periodic", "timeFunc_stepDate",
                      "timeFunc_exp", "timeFunc_log",
                      "timeFunc_uncertaintyQuantification", "timeFunc_timeSeriesCovFile",
                      "timeFunc_bootstrapCount"]},
          {"label": "Geocode",
           "fields": ["geocode", "geocode_SNWE", "geocode_laloStep", "geocode_interpMethod", "geocode_fillValue"]},
          {"label": "Google earth",
           "fields": ["save_kmz"]},
          {"label": "Hdfeos5",
           "fields": ["save_hdfEos5", "save_hdfEos5_update", "save_hdfEos5_subset"]},
          {"label": "Plot",
           "fields": ["plot", "plot_dpi", "plot_maxMemory"]},
          {"label": "HPC (SLURM)",
           "fields": ["hpc_mode"]},
          {"label": "Container",
           "fields": ["container"]},
      ]
      _ui_fields: ClassVar[dict] = {
          # Compute Resources
          "compute_maxMemory":   {"type": "number", "min": 1, "max": 512, "step": 1,
                                  "default": max(1, _env['memory'] - 1),
                                  "hint": "Max memory in GB to allocate (default: system memory minus 1 GB reserve)"},
          "compute_cluster":     {"type": "select",
                                  "options": ["local", "slurm", "pbs", "lsf", "oar", "sge", "none"],
                                  "hint": "Cluster type for parallel processing (local = dask LocalCluster)"},
          "compute_numWorker":   {"type": "number", "min": 1, "max": 64, "step": 1,
                                  "default": _env['cpu'],
                                  "hint": "Number of workers for parallel processing"},
          "compute_config":      {"type": "text",
                                  "hint": "Configuration file for dask distributed cluster"},
          # Load Data
          "load_processor":      {"type": "select",
                                  "options": ["auto", "isce", "aria", "hyp3", "gmtsar", "snap", "gamma", "roipac"],
                                  "hint": "SAR processor of the input dataset"},
          "load_autoPath":       {"type": "text",
                                  "hint": "Auto-detect input file paths based on processor type (auto)"},
          "load_updateMode":     {"type": "select", "options": ["auto", "yes", "no"],
                                  "hint": "Skip re-loading if file already exists with same dataset and metadata"},
          "load_compression":    {"type": "select", "options": ["auto", "lzf", "gzip", "no"],
                                  "hint": "Data compression for HDF5 files"},
          "load_metaFile":       {"type": "text",
                                  "hint": "Metadata file path (ISCE only), e.g. reference/IW1.xml"},
          "load_baselineDir":    {"type": "text",
                                  "hint": "Baseline directory (ISCE only), e.g. baselines"},
          "load_unwFile":        {"type": "text",
                                  "hint": "Unwrapped interferogram file(s), e.g. ./../pairs/*/filt*.unw"},
          "load_corFile":        {"type": "text",
                                  "hint": "Coherence file(s), e.g. ./../pairs/*/filt*.cor"},
          "load_connCompFile":   {"type": "text",
                                  "hint": "Connected components file(s), e.g. ./../pairs/*/filt*.unw.conncomp"},
          "load_intFile":        {"type": "text",
                                  "hint": "Wrapped interferogram file(s), e.g. ./../pairs/*/filt*.int"},
          "load_magFile":        {"type": "text",
                                  "hint": "Interferogram magnitude file(s), e.g. ./../pairs/*/filt*.int"},
          "load_ionUnwFile":     {"type": "text", "hint": "Unwrapped ionospheric phase file(s)"},
          "load_ionCorFile":     {"type": "text", "hint": "Ionospheric coherence file(s)"},
          "load_ionConnCompFile":{"type": "text", "hint": "Ionospheric connected component file(s)"},
          "load_azOffFile":      {"type": "text", "hint": "Azimuth offset file(s)"},
          "load_rgOffFile":      {"type": "text", "hint": "Range offset file(s)"},
          "load_azOffStdFile":   {"type": "text", "hint": "Azimuth offset standard deviation file(s)"},
          "load_rgOffStdFile":   {"type": "text", "hint": "Range offset standard deviation file(s)"},
          "load_offSnrFile":     {"type": "text", "hint": "Offset SNR file(s)"},
          "load_demFile":        {"type": "text",
                                  "hint": "DEM file in radar/geo coordinates, e.g. ./inputs/geometryRadar.h5"},
          "load_lookupYFile":    {"type": "text",
                                  "hint": "Lookup table lat/y file, e.g. ./inputs/geometryGeo.h5"},
          "load_lookupXFile":    {"type": "text", "hint": "Lookup table lon/x file"},
          "load_incAngleFile":   {"type": "text", "hint": "Incidence angle file"},
          "load_azAngleFile":    {"type": "text", "hint": "Azimuth angle file"},
          "load_shadowMaskFile": {"type": "text", "hint": "Shadow/layover mask file"},
          "load_waterMaskFile":  {"type": "text", "hint": "Water mask file"},
          "load_bperpFile":      {"type": "text", "hint": "Perpendicular baseline file"},
          "subset_yx":           {"type": "text", "hint": "Subset in row/column, e.g. 1200:2000,0:2000"},
          "subset_lalo":         {"type": "text", "hint": "Subset in lat/lon, e.g. 37.5:38.5,-118.5:-117.5"},
          "multilook_method":    {"type": "select", "options": ["auto", "mean", "nearest", "no"],
                                  "hint": "Multilook method: mean, nearest, or no for skip"},
          "multilook_ystep":     {"type": "auto_number", "hint": "Multilook factor in y/azimuth direction"},
          "multilook_xstep":     {"type": "auto_number", "hint": "Multilook factor in x/range direction"},
          # Modify Network
          "network_tempBaseMax":     {"type": "auto_number", "hint": "Maximum temporal baseline in days"},
          "network_perpBaseMax":     {"type": "auto_number", "hint": "Maximum perpendicular baseline in meters"},
          "network_connNumMax":      {"type": "auto_number", "hint": "Maximum number of nearest-neighbor connections"},
          "network_startDate":       {"type": "text", "hint": "Start date in YYYYMMDD format"},
          "network_endDate":         {"type": "text", "hint": "End date in YYYYMMDD format"},
          "network_excludeDate":     {"type": "text", "hint": "Date(s) to exclude in YYYYMMDD, separated by space"},
          "network_excludeDate12":   {"type": "text",
                                      "hint": "Interferogram date pairs to exclude, e.g. 20150115_20150127"},
          "network_excludeIfgIndex": {"type": "text",
                                      "hint": "Index(es) of interferograms to exclude, e.g. 2 8 230"},
          "network_referenceFile":   {"type": "text",
                                      "hint": "Reference network file (pairs in date12_list.txt format)"},
          "network_coherenceBased":  {"type": "select", "options": ["auto", "yes", "no"],
                                      "hint": "Enable coherence-based network modification"},
          "network_minCoherence":    {"type": "adaptive_number", "min": 0, "max": 1, "step": 0.05,
                                      "hint": "Minimum coherence for coherence-based modification. "
                                              "'adaptive' = derive from this stack; 'auto' = MintPy 0.7"},
          "network_areaRatioBased":  {"type": "select", "options": ["auto", "yes", "no"],
                                      "hint": "Enable area-ratio-based network modification (ECR method)"},
          "network_minAreaRatio":    {"type": "auto_number",
                                      "hint": "Minimum area ratio for area-ratio-based modification"},
          "network_keepMinSpanTree": {"type": "select", "options": ["auto", "yes", "no"],
                                      "hint": "Keep the minimum spanning tree of the network"},
          "network_maskFile":        {"type": "text",
                                      "hint": "Mask file for coherence-based network modification"},
          "network_aoiYX":           {"type": "text",
                                      "hint": "AOI in row/column for coherence calculation, e.g. 100:200,300:400"},
          "network_aoiLALO":         {"type": "text",
                                      "hint": "AOI in lat/lon for coherence calculation, e.g. 37.5:38.0,-118.0:-117.5"},
          # Reference Point
          "reference_yx":            {"type": "text", "hint": "Reference point in row/column, e.g. 257 151"},
          "reference_lalo":          {"type": "text", "hint": "Reference point in lat/lon, e.g. 37.65 -118.45"},
          "reference_maskFile":      {"type": "text", "hint": "Mask file for reference point selection"},
          "reference_coherenceFile": {"type": "text", "hint": "Coherence file for reference point selection"},
          "reference_minCoherence":  {"type": "adaptive_number", "min": 0, "max": 1, "step": 0.05,
                                      "hint": "Minimum coherence for reference point selection. "
                                              "'adaptive' = derive from this stack; 'auto' = MintPy 0.85"},
          # Unwrap Error
          "unwrapError_method":          {"type": "select",
                                          "options": ["auto", "bridging", "phase_closure",
                                                      "bridging+phase_closure", "no"],
                                          "hint": "Phase unwrapping error correction method"},
          "unwrapError_waterMaskFile":   {"type": "text", "hint": "Water mask file for bridging method"},
          "unwrapError_connCompMinArea": {"type": "auto_number",
                                          "hint": "Minimum area in pixels for a connected component"},
          "unwrapError_numSample":       {"type": "auto_number",
                                          "hint": "Number of randomly sampled triplets for phase_closure method"},
          "unwrapError_ramp":            {"type": "select", "options": ["auto", "linear", "quadratic", "no"],
                                          "hint": "Remove ramp before bridging"},
          "unwrapError_bridgePtsRadius": {"type": "auto_number",
                                          "hint": "Radius in pixels to search for bridge points"},
          # Network Inversion
          "networkInversion_weightFunc":      {"type": "select", "options": ["auto", "var", "fim", "no"],
                                               "hint": "var = spatial variance, fim = Fisher info matrix, no = uniform"},
          "networkInversion_waterMaskFile":   {"type": "text", "hint": "Water mask file applied before inversion"},
          "networkInversion_minNormVelocity": {"type": "select", "options": ["auto", "yes", "no"],
                                               "hint": "Minimize L2-norm of velocity (vs. timeseries) in SBAS inversion"},
          "networkInversion_maskDataset":     {"type": "text",
                                               "hint": "Dataset for masking, e.g. coherence or connectComponent"},
          "networkInversion_maskThreshold":   {"type": "adaptive_number", "min": 0, "max": 1, "step": 0.05,
                                               "hint": "Threshold for maskDataset to mask unwrapped phase. "
                                                       "'adaptive' = derive from this stack; 'auto' = MintPy 0.4"},
          "networkInversion_minRedundancy":   {"type": "auto_number",
                                               "hint": "Minimum redundancy of interferograms per pixel"},
          "networkInversion_minTempCoh":      {"type": "auto_number",
                                               "hint": "Minimum temporal coherence for pixel masking"},
          "networkInversion_minNumPixel":     {"type": "auto_number",
                                               "hint": "Minimum number of coherent pixels to proceed"},
          "networkInversion_shadowMask":      {"type": "select", "options": ["auto", "yes", "no"],
                                               "hint": "Use shadow mask from geometry"},
          # Solid Earth Tides
          "solidEarthTides":  {"type": "select", "options": ["auto", "yes", "no"],
                               "hint": "Correct for solid earth tides using pysolid"},
          # Ionosphere
          "ionosphericDelay_method":       {"type": "select", "options": ["auto", "split_spectrum", "no"],
                                            "hint": "Ionospheric delay correction method"},
          "ionosphericDelay_excludeDate":  {"type": "text",
                                            "hint": "Dates to exclude from ionospheric correction, e.g. 20180202 20180414"},
          "ionosphericDelay_excludeDate12":{"type": "text",
                                            "hint": "Interferogram date pairs to exclude from ionospheric correction"},
          # Troposphere
          "troposphericDelay_method":         {"type": "select",
                                               "options": ["auto", "pyaps", "gacos", "height_correlation", "no"],
                                               "hint": "Tropospheric delay correction method"},
          "troposphericDelay_weatherModel":   {"type": "select",
                                               "options": ["auto", "ERA5", "ERA5T", "MERRA", "NARR"],
                                               "hint": "Weather model for pyaps (ERA5 recommended)"},
          "troposphericDelay_weatherDir":     {"type": "text",
                                               "hint": "Directory of downloaded weather data files for pyaps"},
          "troposphericDelay_polyOrder":      {"type": "auto_number",
                                               "hint": "Polynomial order for height-correlation method"},
          "troposphericDelay_looks":          {"type": "auto_number",
                                               "hint": "Extra multilook factor for height-correlation estimation"},
          "troposphericDelay_minCorrelation": {"type": "auto_number",
                                               "hint": "Minimum correlation between height and phase"},
          "troposphericDelay_gacosDir":       {"type": "text", "hint": "Directory of GACOS delay files"},
          # Deramp
          "deramp":          {"type": "select", "options": ["auto", "linear", "quadratic", "no"],
                              "hint": "Remove phase ramp in x/y direction"},
          "deramp_maskFile": {"type": "text", "hint": "Mask file for ramp estimation"},
          # Topography
          "topographicResidual":                 {"type": "select", "options": ["auto", "yes", "no"],
                                                  "hint": "Correct topographic residuals (DEM error)"},
          "topographicResidual_polyOrder":       {"type": "auto_number",
                                                  "hint": "Polynomial order for DEM error estimation"},
          "topographicResidual_phaseVelocity":   {"type": "select", "options": ["auto", "yes", "no"],
                                                  "hint": "Minimize phase velocity (not phase) in DEM error inversion"},
          "topographicResidual_stepDate":        {"type": "text",
                                                  "hint": "Step function date(s) for co-seismic jumps, e.g. 20140911"},
          "topographicResidual_excludeDate":     {"type": "text",
                                                  "hint": "Dates to exclude in DEM error inversion"},
          "topographicResidual_pixelwiseGeometry":{"type": "select", "options": ["auto", "yes", "no"],
                                                   "hint": "Use pixel-wise geometry in DEM error estimation"},
          # Residual RMS
          "residualRMS_maskFile": {"type": "text", "hint": "Mask file for residual phase quality assessment"},
          "residualRMS_deramp":   {"type": "select", "options": ["auto", "linear", "quadratic", "no"],
                                   "hint": "Remove ramp before RMS calculation"},
          "residualRMS_cutoff":   {"type": "auto_number",
                                   "hint": "Cutoff value in RMS threshold for outlier date detection"},
          # Reference Date
          "reference_date": {"type": "text",
                             "hint": "Reference date in YYYYMMDD; 'auto' = first date with full coherence"},
          # Velocity
          "timeFunc_startDate":                {"type": "text", "hint": "Start date of the time function fit"},
          "timeFunc_endDate":                  {"type": "text", "hint": "End date of the time function fit"},
          "timeFunc_excludeDate":              {"type": "text",
                                                "hint": "Date(s) to exclude from time function fitting"},
          "timeFunc_polynomial":               {"type": "auto_number",
                                                "hint": "Polynomial order: 1 = linear velocity, 2 = acceleration"},
          "timeFunc_periodic":                 {"type": "text",
                                                "hint": "Periodic periods in years, e.g. 1.0 0.5 for annual+semi-annual"},
          "timeFunc_stepDate":                 {"type": "text",
                                                "hint": "Step function date(s), e.g. 20161231 for co-seismic jump"},
          "timeFunc_exp":                      {"type": "text",
                                                "hint": "Exponential decay: onset_date char_time, e.g. 20181026 60"},
          "timeFunc_log":                      {"type": "text",
                                                "hint": "Logarithmic relaxation: onset_date char_time, e.g. 20181026 60"},
          "timeFunc_uncertaintyQuantification":{"type": "select", "options": ["auto", "bootstrap", "residue"],
                                                "hint": "Method for velocity uncertainty quantification"},
          "timeFunc_timeSeriesCovFile":        {"type": "text",
                                                "hint": "Time-series covariance file for uncertainty propagation"},
          "timeFunc_bootstrapCount":           {"type": "auto_number",
                                                "hint": "Number of bootstrap iterations"},
          # Geocode
          "geocode":              {"type": "select", "options": ["auto", "yes", "no"],
                                   "hint": "Geocode datasets in radar coordinates to geo coordinates"},
          "geocode_SNWE":         {"type": "text",
                                   "hint": "Bounding box: south north west east, e.g. 31 40 -115 -100"},
          "geocode_laloStep":     {"type": "text",
                                   "hint": "Output pixel size in lat/lon, e.g. -0.000833 0.000833 (≈90 m)"},
          "geocode_interpMethod": {"type": "select", "options": ["auto", "nearest", "linear"],
                                   "hint": "Interpolation method for geocoding"},
          "geocode_fillValue":    {"type": "text",
                                   "hint": "Fill value for pixels outside coverage, e.g. nan or 0"},
          # Google Earth
          "save_kmz":            {"type": "select", "options": ["auto", "yes", "no"],
                                  "hint": "Save geocoded velocity to Google Earth KMZ file"},
          # HDF-EOS5
          "save_hdfEos5":        {"type": "select", "options": ["auto", "yes", "no"],
                                  "hint": "Save time-series to HDF-EOS5 format"},
          "save_hdfEos5_update": {"type": "select", "options": ["auto", "yes", "no"],
                                  "hint": "Update HDF-EOS5 file if already exists"},
          "save_hdfEos5_subset": {"type": "select", "options": ["auto", "yes", "no"],
                                  "hint": "Save subset of HDF-EOS5 file"},
          # Plot
          "plot":                {"type": "select", "options": ["auto", "yes", "no"],
                                  "hint": "Plot results during processing"},
          "plot_dpi":            {"type": "auto_number", "hint": "Figure DPI for saved plots"},
          "plot_maxMemory":      {"type": "auto_number",
                                  "hint": "Maximum memory in GB for plot_smallbaseline.py"},
          "hpc_mode":            {"type": "bool",
                                  "hint": "Submit the full MintPy run as a single sbatch job. "
                                          "SLURM resources come from sbatch_options.json (step \"17\": \"SBAS\") "
                                          "in the workdir, generated automatically on first use."},
          "container":           {"type": "text",
                                  "hint": "Path to a .sif/Apptainer image or a Docker image reference with insarhub "
                                          "installed — re-runs this command inside the container instead of on the "
                                          "host. Not remembered between runs; pass again for subsequent runs."},
      }
      # ─────────────────────────────────────────────────────────────────────────
  
      name: str = "Mintpy_SBAS_Base_Config"
      workdir: Path | str = field(default_factory=lambda: Path.cwd())
      debug: bool = False
      hpc_mode: bool = False
      container: str | None = None
      # Default container image used when `--container` is passed with no value.
      # MintPy analyzers need MintPy + (for ISCE2) ISCE2 -- the isce2 image has both.
      container_default: str = "ghcr.io/jldz9/insarhub-isce2-mintpy:0.4.0"
  
      ## computing resource configuration
      # System memory minus a 1 GB reserve for the OS/scheduler: giving dask the
      # FULL machine RAM over-subscribes every worker and OOM-kills them
      # ("Lost all workers"). 31 G -> 30 G on a 32 G box.
      compute_maxMemory : float | int = max(1, _env['memory'] - 1)
      compute_cluster : str = 'local' # Mintpy's slurm parallel processing is buggy, so we will handle parallel processing with dask instead. Switch to none to turn off parallel processing to save memory.
      compute_numWorker : int = _env['cpu']
      compute_config: str = 'none'
  
      ## Load data
      load_processor: str = 'auto'
      load_autoPath: str = 'auto' 
      load_updateMode: str = 'no'
      load_compression: str = 'auto'
      ##---------for ISCE only:
      load_metaFile: str = 'auto'
      load_baselineDir: str = 'auto'
      ##---------interferogram stack:
      load_unwFile: str = 'auto'
      load_corFile: str = 'auto'
      load_connCompFile: str = 'auto'
      load_intFile: str = 'auto'
      load_magFile: str = 'auto'
      ##---------ionosphere stack (optional):
      load_ionUnwFile: str = 'auto'
      load_ionCorFile: str = 'auto'
      load_ionConnCompFile: str = 'auto'
      ##---------offset stack (optional):
      load_azOffFile: str = 'auto'
      load_rgOffFile: str = 'auto'
      load_azOffStdFile: str = 'auto'
      load_rgOffStdFile: str = 'auto'
      load_offSnrFile: str = 'auto'
      ##---------geometry:
      load_demFile: str = 'auto'
      load_lookupYFile: str = 'auto'
      load_lookupXFile: str = 'auto'
      load_incAngleFile: str = 'auto'
      load_azAngleFile: str = 'auto'
      load_shadowMaskFile: str = 'auto'
      load_waterMaskFile: str = 'auto'
      load_bperpFile: str = 'auto'
      ##---------subset (optional):
      subset_yx: str = 'auto'
      subset_lalo: str = 'auto'
      ##---------multilook (optional):
      multilook_method: str = 'auto'
      multilook_ystep: str | int = 'auto'
      multilook_xstep: str | int= 'auto'
  
      # 2. Modify Network
      network_tempBaseMax: str | float = 'auto'
      network_perpBaseMax: str | float = 'auto'
      network_connNumMax: str | int = 'auto'
      network_startDate: str = 'auto'
      network_endDate: str = 'auto'
      network_excludeDate: str = 'auto'
      network_excludeDate12: str = 'auto'
      network_excludeIfgIndex: str = 'auto'
      network_referenceFile: str = 'auto'
      ## 2) Data-driven network modification
      network_coherenceBased: str = 'auto'
      network_minCoherence: str |float = 'auto'
      ## b - Effective Coherence Ratio network modification = (threshold + MST) by default
      network_areaRatioBased: str = 'auto'
      network_minAreaRatio: str |float= 'auto'
      ## Additional common parameters for the 2) data-driven network modification
      network_keepMinSpanTree: str = 'auto'
      network_maskFile: str = 'auto'
      network_aoiYX: str = 'auto'
      network_aoiLALO: str = 'auto'
  
      # 3. Reference Point
      reference_yx: str = 'auto'
      reference_lalo: str = 'auto'
      reference_maskFile: str = 'auto'
      reference_coherenceFile: str = 'auto'
      reference_minCoherence: str |float = 'auto'
  
      # 4. Correct Unwrap Error
      unwrapError_method: str = 'auto'
      unwrapError_waterMaskFile: str = 'auto'
      unwrapError_connCompMinArea: str |float = 'auto'
      ## phase_closure options:
      unwrapError_numSample: str | int= 'auto'
      ## bridging options:
      unwrapError_ramp: str = 'auto'
      unwrapError_bridgePtsRadius: str | int= 'auto'
  
      # 5. Invert Network
      networkInversion_weightFunc: str = 'auto'
      networkInversion_waterMaskFile: str = 'auto'
      networkInversion_minNormVelocity: str = 'auto'
      ## mask options for unwrapPhase of each interferogram before inversion (recommend if weightFunct=no):
      networkInversion_maskDataset: str = 'auto'
      networkInversion_maskThreshold: str | float = 'auto'
      networkInversion_minRedundancy: str | float = 'auto'
      ## Temporal coherence is calculated and used to generate the mask as the reliability measure
      networkInversion_minTempCoh: str | float = 'auto'
      networkInversion_minNumPixel: str | int = 'auto'
      networkInversion_shadowMask: str = 'auto'
  
      # 6. Correct SET (Solid Earth Tides)
      solidEarthTides: str = 'auto'
  
      # 7. Correct Ionosphere
      ionosphericDelay_method: str = 'auto'
      ionosphericDelay_excludeDate: str = 'auto'
      ionosphericDelay_excludeDate12: str = 'auto'
  
      # 8. Correct Troposphere
      troposphericDelay_method: str = 'auto'
      ## Notes for pyaps:
      troposphericDelay_weatherModel: str = 'auto'
      troposphericDelay_weatherDir: str = 'auto'
  
      ## Notes for height_correlation:
      troposphericDelay_polyOrder: str | int = 'auto'
      troposphericDelay_looks: str | int = 'auto'
      troposphericDelay_minCorrelation: str | float = 'auto'
      ## Notes for gacos:
      troposphericDelay_gacosDir: str = 'auto'
  
      # 9. Deramp
      deramp: str = 'auto'
      deramp_maskFile: str = 'auto'
  
      # 10. Correct Topography
      topographicResidual: str = 'auto'
      topographicResidual_polyOrder: str = 'auto'
      topographicResidual_phaseVelocity: str = 'auto'
      topographicResidual_stepDate: str = 'auto'
      topographicResidual_excludeDate: str = 'auto'
      topographicResidual_pixelwiseGeometry: str = 'auto'
  
      # 11.1 Residual RMS
      residualRMS_maskFile: str = 'auto'
      residualRMS_deramp: str = 'auto'
      residualRMS_cutoff: str | float = 'auto'
  
      # 11.2 Reference Date
      reference_date: str = 'auto'
  
      # 12. Velocity
      timeFunc_startDate: str = 'auto'
      timeFunc_endDate: str = 'auto'
      timeFunc_excludeDate: str = 'auto'
      ## Fit a suite of time functions
      timeFunc_polynomial: str | int = 'auto'
      timeFunc_periodic: str = 'auto'
      timeFunc_stepDate: str = 'auto'
      timeFunc_exp: str = 'auto'
      timeFunc_log: str = 'auto'
      ## Uncertainty quantification methods:
      timeFunc_uncertaintyQuantification: str = 'auto'
      timeFunc_timeSeriesCovFile: str = 'auto'
      timeFunc_bootstrapCount: str | int = 'auto'
  
      # 13.1 Geocode
      geocode: str = 'auto'
      geocode_SNWE: str = 'auto'
      geocode_laloStep: str = 'auto'
      geocode_interpMethod: str = 'auto'
      geocode_fillValue: str | float = 'auto'
  
      # 13.2 Google Earth
      save_kmz: str = 'auto'
  
      # 13.3 HDFEOS5
      save_hdfEos5: str = 'auto'
      save_hdfEos5_update: str = 'auto'
      save_hdfEos5_subset: str = 'auto'
  
      # 13.4 Plot
      plot: str = 'auto'
      plot_dpi: str | int = 'auto'
      plot_maxMemory: str | int = 'auto'
  
      def __post_init__(self):
          if isinstance(self.workdir, str):
              self.workdir = Path(self.workdir).expanduser().resolve()
  
      def write_mintpy_config(self, outpath: Union[Path, str]):
          """
          Writes the dataclass to a mintpy .cfg file, excluding operational 
          parameters that MintPy doesn't recognize.
          """
          outpath = Path(outpath).expanduser().resolve()
          outpath.parent.mkdir(parents=True, exist_ok=True)
          exclude_fields = ['name', 'workdir', 'debug']
          # InSARHub stores these space-separated (e.g. "37.84 -112.82",
          # matching --reference_lalo CLI input), but MintPy's own template
          # reader does value.split(',') -- it requires "lat,lon"/"y,x".
          comma_join_fields = ['reference_yx', 'reference_lalo']
  
          with open(outpath, 'w') as f:
              f.write("## MintPy Config File Generated via InSARHub\n")
  
              for key, value in asdict(self).items():
                  if key in exclude_fields:
                      continue
  
                  if key in comma_join_fields and isinstance(value, str) and ',' not in value:
                      parts_val = value.split()
                      if len(parts_val) == 2:
                          value = ",".join(parts_val)
  
                  parts = key.split('_')
                  if len(parts) > 1:
                      mintpy_key = f"mintpy.{parts[0]}.{'.'.join(parts[1:])}"
                  else:
                      mintpy_key = f"mintpy.{parts[0]}"
  
                  f.write(f"{mintpy_key:<40} = {value}\n")
  
          return Path(outpath).resolve()
  

  Adaptive coherence thresholds

  Three coherence parameters default to the literal "adaptive" instead of a fixed number. During prep_data, InSARHub inspects the stack's actual coherence distribution and resolves each into .mintpy.cfg:

  Parameter	Resolved from	Cap
  network_minCoherence	strictest threshold that keeps the network connected + redundant	≤ 0.6
  networkInversion_maskThreshold	percentile that keeps the reliable fraction of pixels	≤ 0.6
  reference_minCoherence	98th percentile (min 0.30) for a stable reference point	≤ 0.85

  Adaptation only kicks in when the data is below the cap; a clean, high-coherence stack simply gets the cap value. Set any of these to an explicit number to override the adaptive logic entirely.

• Run

  Run the Mintpy time-series analysis based on provided configuration

  analyzer.run()
  

  Parameters:

  Name	Type	Description	Default
  steps	list[str] | None	List of MintPy processing steps to execute. If None, the default full workflow is executed: [ 'load_data', 'modify_network', 'reference_point', 'quick_overview', 'correct_unwrap_error', 'invert_network', 'correct_LOD', 'correct_SET', 'correct_ionosphere', 'correct_troposphere', 'deramp', 'correct_topography', 'residual_RMS', 'reference_date', 'velocity', 'geocode', 'google_earth', 'hdfeos5' ]	None

  Raises:

  Type	Description
  RuntimeError	If tropospheric delay method requires CDS authorization and authorization fails.
  Exception	Propagates exceptions raised during MintPy execution.

• Submit (HPC / SLURM mode)

  Generate a single sbatch script covering all selected steps and submit it to SLURM. Inherited by Hyp3_Mintpy_SBAS and ISCE2_Mintpy_SBAS.

  # Submit full pipeline as one SLURM job
  analyzer.submit_hpc()
  
  # Submit only specific steps
  analyzer.submit_hpc(steps=["velocity", "geocode"])
  

  The script is written to <workdir>/mintpy/mintpy_sbas.sbatch and job state to mintpy/mintpy_job.json. SLURM resources come from <workdir>/sbatch_options.json, step key "17" — the same file ISCE2_S1's own HPC submission uses for steps 01–16, since the processor and analyzer typically share one workdir. Default: time=24:00:00, ntasks=1, cpus_per_task=16, mem=128G, partition=all.

  submit_hpc() returns the SLURM job ID string on success, or None if sbatch_options.json was just created (or updated with a missing "17" entry) — callers should check for None and stop rather than treat it as a successful submission:

  cfg = Mintpy_SBAS_Base_Config(
      workdir="/your/work/dir",
      load_processor="hyp3",
      hpc_mode=True,
  )
  analyzer = Analyzer.create('Hyp3_Mintpy_SBAS', config=cfg)
  job_id = analyzer.submit_hpc()
  if job_id is None:
      print("sbatch_options.json was just created/updated — review it, then resubmit.")
  

  Edit step "17" in sbatch_options.json directly to change resources (e.g. {"17": {"time": "48:00:00", "mem": "256G", "partition": "gpu"}}), then call submit_hpc() again.

• Plot

  (Re)generate the figures under mintpy/pic/ from already-computed results, without recomputing anything. run()'s own auto-plot only fires for a single call covering more than one step (mirroring MintPy's own CLI semantics) — the CLI and GUI execute steps one at a time internally for per-step progress reporting, so that condition never actually fires there; plot() is the explicit, standalone alternative both call once after their step sequence completes (or on demand, e.g. after tweaking a plotting-related config value and wanting fresh figures without rerunning the whole pipeline).

  analyzer.plot()
  

• Running without a local MintPy (or ISCE2) install

  Set the container field to a path to an Apptainer/Singularity .sif image, or a Docker image reference (name[:tag]), and run()/prep_data()/submit_hpc() all re-invoke the same insarhub analyzer ... CLI call inside that container instead of on the host — the workdir is bind-mounted at the identical path, so output lands exactly where a native run would put it. The container image just needs insarhub installed alongside MintPy (and ISCE2, for ISCE2_Mintpy_SBAS) — see docker/dev/Dockerfile.isce2-mintpy for a ready-to-build example.

  cfg = Mintpy_SBAS_Base_Config(
      workdir="/your/work/dir",
      load_processor="hyp3",
      container="ghcr.io/jldz9/insarhub-isce2-mintpy:0.4.0",
  )
  analyzer = Analyzer.create('Hyp3_Mintpy_SBAS', config=cfg)
  analyzer.run()
  

  container is a per-invocation setting, not persisted config — it must be set again on every subsequent call that should also run inside the container.

• Clean up

  Remove intermediate processing files generated during the time-series process

  analyzer.cleanup()
  

  Raises:

  Type	Description
  Exception	Propagates any unexpected errors raised during removal.

The Hyp3_Mintpy_SBAS is a specialized analyzer that extends Mintpy_SBAS_Base_Analyzer, preconfigured specifically for processing time-series data from HyP3 InSAR products.

Source code in src/insarhub/analyzer/hyp3_mintpy_s1_sbas.py

class Hyp3_Mintpy_SBAS(Mintpy_SBAS_Base_Analyzer):
    name = 'Hyp3_Mintpy_SBAS'
    aliases = ('Hyp3_SBAS', 'Hyp3_TS', 'Hyp3_Mintpy_TS')   # legacy names
    description = "SBAS time-series analysis of HyP3 InSAR outputs using MintPy."
    compatible_processor = "Hyp3_S1"
    default_config = Hyp3_Mintpy_SBAS_Config
    required = ['unw_phase.tif', 'corr.tif',  'dem.tif'] # also need meta files to get the date and other info
    optional = ['lv_theta.tif', 'lv_phi.tif', 'water_mask.tif']
    # own output dir (workdir/hyp3_mintpy/) -- keeps this analyzer's MintPy
    # products separate from ISCE2_Mintpy_SBAS/GMTSAR_Mintpy_SBAS runs on the same
    # workdir (previously all shared workdir/mintpy/ and silently overwrote
    # each other). Path layout centralized in config/paths.py (MintPyPaths).
    MINTPY_SUBDIR = "hyp3_mintpy"

    def __init__(self, config: Hyp3_Mintpy_SBAS_Config | None = None):
        super().__init__(config)

    def prep_data(self):
        """
        Prepare input data for analysis by performing unzipping, collection, clipping, and parameter setup.

        This method orchestrates the preprocessing steps required before running the analysis workflow. 
        It ensures that all input files are available, aligned, and properly configured.

        Steps performed:
            1. `_unzip_hyp3()`: Extracts any compressed Hyp3 output files.
            2. `_collect_files()`: Gathers relevant input files (e.g., DEMs, interferograms).
            3. `_get_common_overlap(files['dem'])`: Computes the spatial overlap extent among input rasters.
            4. `_clip_rasters(files, overlap_extent)`: Clips input rasters to the common overlapping area.
            5. `_set_load_parameters()`: Sets parameters required for loading the preprocessed data into memory.

        Raises:
            FileNotFoundError: If required input files are missing.
            ValueError: If no common overlap region can be determined among rasters.
            Exception: Propagates any unexpected errors during preprocessing.

        Notes:
            - This method must be called before running the analysis workflow.
            - Designed for workflows using Hyp3-derived Sentinel-1 products.
            - Ensures consistent spatial coverage across all input datasets.
        """
        if self.config.container:
            return self._run_via_container(["prep_data"])

        self._unzip_hyp3()
        files = self._collect_files()
        overlap_extent = self._get_common_overlap(files['dem'])
        self._clip_rasters(files, overlap_extent)
        self._set_load_parameters()
        super().prep_data()

    def _unzip_hyp3(self):
        print(f'{Fore.CYAN}Unzipping HyP3 Products...{Fore.RESET}')

        hyp3_dir = self._hyp3_paths.output_dir
        search_root = hyp3_dir if hyp3_dir.exists() else Path(self.workdir)
        hyp3_results = list(search_root.rglob('*.zip'))
        self.tmp_dir.mkdir(parents=True, exist_ok=True)

        with tqdm(hyp3_results, desc="Processing", unit="file") as pbar:
            for zip_file in pbar:
                extract_target = self.tmp_dir / zip_file.stem
                with zipfile.ZipFile(zip_file, 'r') as zf:
                    needs_extraction = True
                    if extract_target.is_dir():
                        files_in_zip = {Path(f).name for f in zf.namelist() if not f.endswith('/')}
                        folder_files = {f.name for f in extract_target.iterdir() if f.is_file()}
                        if files_in_zip.issubset(folder_files):
                            needs_extraction = False
                            pbar.set_description(f"File Exist: {zip_file.stem[:30]}...")
                    if needs_extraction:
                        pbar.set_description(f"Extracting: {zip_file.stem[:30]}...")
                        if extract_target.is_dir():
                            shutil.rmtree(extract_target)

                        zf.extractall(self.tmp_dir)
        print(f'\n{Fore.GREEN}Unzipping complete.{Fore.RESET}')

    def _collect_files(self):
        print(f'{Fore.CYAN}Mapping file paths...{Fore.RESET}')
        all_required = {ext.split('.')[0]: ext for ext in self.required}    
        all_optional = {ext.split('.')[0]: ext for ext in self.optional}
        files = defaultdict(list)
        files['meta'] = [m for m in self.tmp_dir.rglob('*.txt') if 'README' not in m.name]
        for cat_name, ext in {**all_required, **all_optional}.items():
            files[cat_name] = list(self.tmp_dir.rglob(f"*_{ext}"))

        missing_req = [name for name, ext in all_required.items() if not files[name]]
        if missing_req or not files['meta']:
            print(f"\033[K", end="\r") # Clear current line
            msg = []
            if missing_req: msg.append(f"Missing rasters: {missing_req}")
            if not files['meta']: msg.append("Missing metadata (.txt) files")

            error_report = f"{Fore.RED}CRITICAL ERROR: {'. '.join(msg)}.{Fore.RESET}\n" \
                           f"MintPy requires these files to extract dates and baselines."
            raise FileNotFoundError(error_report)
        missing_opt = [name for name in all_optional if not files[name]]

        total_pairs = len(files['unw_phase'])
        status_msg = f"{Fore.GREEN}Found {total_pairs} pairs | Metadata: OK"
        if missing_opt:
            status_msg += f" | {Fore.YELLOW}Missing optional: {missing_opt}"

        print(f"\r\033[K{status_msg}{Fore.RESET}")
        return files

    def _get_common_overlap(self, dem_files):
        import rasterio
        lefts, bottoms, rights, tops = [], [], [], []
        for f in dem_files:
            with rasterio.open(f.as_posix()) as ds:
                b = ds.bounds
            lefts.append(b.left)
            bottoms.append(b.bottom)
            rights.append(b.right)
            tops.append(b.top)
        # (left, top, right, bottom) of the intersection across all rasters
        return (max(lefts), min(tops), min(rights), max(bottoms))

    def _clip_rasters(self, files, overlap_extent):
        import rasterio
        from rasterio.windows import from_bounds

        print(f'{Fore.CYAN}Clipping rasters to common overlap...{Fore.RESET}')
        self.clip_dir.mkdir(parents=True, exist_ok=True)
        categories = [k for k in files.keys() if k != 'meta']
        left, top, right, bottom = overlap_extent

        def _is_valid_raster(path: Path) -> bool:
            """Check that path is a real, fully-written raster (not left behind
            by an interrupted prior clip -- crash, Ctrl+C, disk full, etc.)."""
            try:
                with rasterio.open(path.as_posix()) as ds:
                    return ds.count > 0
            except Exception:
                return False

        with tqdm(categories, desc="Progress", position=0, dynamic_ncols=True) as pbar_out:
            for key in pbar_out:
                file_list = files[key]
                pbar_out.set_description(f"Group: {key}")

                # Inner progress bar for individual files in this group
                # leave=False ensures the inner bar disappears when the group is done
                with tqdm(file_list, desc=f"  -> Clipping", leave=False, position=1, unit="file", dynamic_ncols=True) as pbar_in:
                    for f in pbar_in:
                        out = self.clip_dir / f"{f.stem}_clip.tif"

                        if out.exists():
                            if _is_valid_raster(out):
                                pbar_in.set_postfix_str(f"Skip: {f.name[:15]}...")
                                # Update postfix instead of printing to avoid creating new lines
                                continue
                            # Left behind by an interrupted prior run -- existing but
                            # broken, so a plain exists()-check would skip it forever.
                            tqdm.write(f"{Fore.YELLOW}  {out.name} exists but isn't a valid "
                                       f"raster (interrupted prior run?) — re-clipping.{Fore.RESET}")
                            out.unlink(missing_ok=True)

                        pbar_in.set_postfix_str(f"File: {f.name[:15]}...")

                        # Write to a temp path and rename on success only, so a
                        # crash/interrupt mid-write never leaves a broken file
                        # sitting at the final path for a future run to skip over.
                        tmp_out = out.parent / (out.name + ".part")
                        try:
                            with rasterio.open(f.as_posix()) as src:
                                window = from_bounds(left, bottom, right, top, transform=src.transform)
                                window = window.round_offsets().round_lengths()
                                transform = src.window_transform(window)
                                data = src.read(window=window)
                                profile = src.profile.copy()
                                profile.update(height=window.height, width=window.width, transform=transform)
                                with rasterio.open(tmp_out.as_posix(), "w", **profile) as dst:
                                    dst.write(data)
                            tmp_out.rename(out)
                        except Exception as e:
                            tmp_out.unlink(missing_ok=True)
                            tqdm.write(f"{Fore.RED}Error clipping {f.name}: {e}{Fore.RESET}")

            # Handle metadata separately as it's just a file copy (no progress bar needed)
        if 'meta' in files:
            print(f"\r{Fore.CYAN}Step: Copying metadata files... \033[K", end="", flush=True)
            for f in files['meta']:
                shutil.copy(f, self.clip_dir / f.name)

        print(f'\n{Fore.GREEN}Clipping complete.{Fore.RESET}')

    def _set_load_parameters(self):
        self.config.load_unwFile = (self.clip_dir / '*_unw_phase_clip.tif').as_posix()
        self.config.load_corFile = (self.clip_dir / '*_corr_clip.tif').as_posix()
        self.config.load_demFile = (self.clip_dir / '*_dem_clip.tif').as_posix()
        opt_map = {
            'lv_theta': 'load_incAngleFile',
            'lv_phi': 'load_azAngleFile',
            'water_mask': 'load_waterMaskFile'
        }
        for k, cfg_attr in opt_map.items():
            if list(self.clip_dir.glob(f"*_{k}_clip.tif")):
                setattr(self.config, cfg_attr, (self.clip_dir / f"*_{k}_clip.tif").as_posix())

Usage

• Create Analyzer with Parameters

  Initialize an analyzer instance

  analyzer = Analyzer.create('Hyp3_Mintpy_SBAS',
                              workdir="/your/work/dir")
  

  OR

  params = {"workdir": "/your/work/dir"}
  analyzer = Analyzer.create('Hyp3_Mintpy_SBAS', **params)
  

  OR

  from insarhub.config import Mintpy_SBAS_Base_Config
  cfg = Mintpy_SBAS_Base_Config(workdir="/your/work/dir")
  analyzer = Analyzer.create('Hyp3_Mintpy_SBAS', config=cfg)
  

• Prepare data

  Prepare interferogram data downloaded from HyP3 server for MintPy

  analyzer.prep_data()
  

  Raises:

  Type	Description
  FileNotFoundError	If required input files are missing.
  ValueError	If no common overlap region can be determined among rasters.
  Exception	Propagates any unexpected errors during preprocessing.

  Source code in src/insarhub/analyzer/hyp3_mintpy_s1_sbas.py

  def prep_data(self):
      """
      Prepare input data for analysis by performing unzipping, collection, clipping, and parameter setup.
  
      This method orchestrates the preprocessing steps required before running the analysis workflow. 
      It ensures that all input files are available, aligned, and properly configured.
  
      Steps performed:
          1. `_unzip_hyp3()`: Extracts any compressed Hyp3 output files.
          2. `_collect_files()`: Gathers relevant input files (e.g., DEMs, interferograms).
          3. `_get_common_overlap(files['dem'])`: Computes the spatial overlap extent among input rasters.
          4. `_clip_rasters(files, overlap_extent)`: Clips input rasters to the common overlapping area.
          5. `_set_load_parameters()`: Sets parameters required for loading the preprocessed data into memory.
  
      Raises:
          FileNotFoundError: If required input files are missing.
          ValueError: If no common overlap region can be determined among rasters.
          Exception: Propagates any unexpected errors during preprocessing.
  
      Notes:
          - This method must be called before running the analysis workflow.
          - Designed for workflows using Hyp3-derived Sentinel-1 products.
          - Ensures consistent spatial coverage across all input datasets.
      """
      if self.config.container:
          return self._run_via_container(["prep_data"])
  
      self._unzip_hyp3()
      files = self._collect_files()
      overlap_extent = self._get_common_overlap(files['dem'])
      self._clip_rasters(files, overlap_extent)
      self._set_load_parameters()
      super().prep_data()
  

• Run

  Run the Mintpy time-series analysis based on provided configuration

  analyzer.run()
  

  Parameters:

  Name	Type	Description	Default
  steps	list[str] | None	List of MintPy processing steps to execute. If None, the default full workflow is executed: [ 'load_data', 'modify_network', 'reference_point', 'quick_overview', 'correct_unwrap_error', 'invert_network', 'correct_LOD', 'correct_SET', 'correct_ionosphere', 'correct_troposphere', 'deramp', 'correct_topography', 'residual_RMS', 'reference_date', 'velocity', 'geocode', 'google_earth', 'hdfeos5' ]	None

  Raises:

  Type	Description
  RuntimeError	If tropospheric delay method requires CDS authorization and authorization fails.
  Exception	Propagates exceptions raised during MintPy execution.

  Source code in src/insarhub/analyzer/mintpy_base.py

  def run(self, steps=None):
      """
      Run the MintPy SBAS time-series analysis workflow.
  
      This method writes the MintPy configuration file, optionally authorizes
      CDS access for tropospheric correction, and executes the selected
      MintPy processing steps using TimeSeriesAnalysis.
  
      Args:
          steps (list[str] | None, optional):
              List of MintPy processing steps to execute. If None, the
              default full workflow is executed:
                  [
                      'load_data', 'modify_network', 'reference_point', 'quick_overview',
                      'correct_unwrap_error', 'invert_network', 'correct_LOD', 'correct_SET',
                      'correct_ionosphere', 'correct_troposphere',
                      'deramp', 'correct_topography', 'residual_RMS',
                      'reference_date', 'velocity', 'geocode',
                      'google_earth', 'hdfeos5'
                  ]
  
      Raises:
          RuntimeError: If tropospheric delay method requires CDS authorization
              and authorization fails.
          Exception: Propagates exceptions raised during MintPy execution.
  
      Notes:
          - If `troposphericDelay_method` is set to 'pyaps', CDS
          authorization is performed before running MintPy.
          - The configuration file is written to `self.cfg_path`.
          - Processing is executed inside `self.workdir`.
          - This method wraps MintPy TimeSeriesAnalysis for SBAS workflows.
      """
      # HPC: hand the whole analysis to SLURM instead of running MintPy in this
      # process -- mirrors processor.submit()'s hpc dispatch so the API is
      # symmetric (set hpc_mode, call run()). Returns submit_hpc()'s job id, or
      # None if it just wrote sbatch_options.json for review (call run() again
      # after tuning it). The sbatch body re-invokes `insarhub analyzer ... run`
      # WITHOUT --hpc-mode (hpc_mode is skipped by _serialize_config_overrides),
      # so the compute-node run() sees hpc_mode=False and runs locally -- no
      # resubmission loop. Guarded off inside a container child for the same
      # reason (hpc_mode isn't carried in there either).
      if getattr(self.config, "hpc_mode", False) and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
          return self.submit_hpc(steps=steps)
  
      # not INSARHUB_CONTAINER_CHILD: run the steps locally when already inside
      # the container (see prep_data's guard for the full rationale).
      if self.config.container and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
          return self._run_via_container(steps)
  
      run_steps = steps or [
          'load_data', 'modify_network', 'reference_point', 'quick_overview',
          'correct_unwrap_error', 'invert_network',
          'correct_LOD', 'correct_SET', 'correct_ionosphere', 'correct_troposphere',
          'deramp', 'correct_topography', 'residual_RMS', 'reference_date',
          'velocity', 'geocode', 'google_earth', 'hdfeos5'
      ]
  
      # prep_data is what fills mintpy.load.* with the real geocoded file
      # paths (plus the resolved adaptive thresholds and HEADING). The GUI
      # lets users deselect it, and load_data can be run on its own, so
      # self-heal: if it isn't in this run and the cfg still has no resolved
      # load paths, run prep_data first. Otherwise MintPy finds no files,
      # writes no ifgramStack.h5, and load_data fails. prep_data is cheap to
      # repeat (cached DEM / baselines).
      if 'prep_data' not in run_steps and not self._cfg_load_paths_resolved():
          print(f"{Fore.YELLOW}mintpy.load.* not resolved yet — running prep_data "
                f"first to set the file locations.{Fore.RESET}")
          self.prep_data()
  
      if not self.cfg_path.exists():
          print(f"{Fore.YELLOW}Warning: .mintpy.cfg not found — writing config now. "
                f"If this is a Hyp3_Mintpy_SBAS run, make sure 'prep_data' (or '--step prep') "
                f"was completed first so load parameters are correct.{Fore.RESET}")
      # Re-apply the (possibly CLI-/GUI-overridden) config to .mintpy.cfg on
      # every run, not just the first: prep_data creates the file, so without
      # this any parameter passed to a later step (e.g. --networkInversion_
      # minTempCoh on invert_network) was silently dropped because the stale
      # file already existed. Preserves the load paths / HEADING prep_data
      # computed into the file (they are not on self.config here).
      self._sync_runtime_cfg()
  
      if self.config.troposphericDelay_method == 'pyaps' and 'correct_troposphere' in run_steps:
          self._cds_authorize()
      print(f'{Style.BRIGHT}{Fore.MAGENTA}Running MintPy Analysis...{Fore.RESET}')
      self.mintpy_dir.mkdir(parents=True, exist_ok=True)
      _patch_mintpy_plot_bugs()
      from mintpy.smallbaselineApp import TimeSeriesAnalysis
      app = TimeSeriesAnalysis(self.cfg_path.as_posix(), self.mintpy_dir.as_posix())
      try:
          app.open()
          app.run(steps=run_steps)
          if 'geocode' in run_steps:
              self._geocode_diagnostic_files(self.mintpy_dir)
          # Mirrors mintpy.smallbaselineApp's own CLI wrapper
          # (run_smallbaselineApp()), which calls these two after run() --
          # plot_result() is what actually populates mintpy_dir/pic/, and
          # close() is what restores the process's working directory after
          # open() changed into mintpy_dir (skipping it would leave a
          # long-running server process permanently cd'd into the last
          # analyzed folder).
          if app.template.get('mintpy.plot') and len(run_steps) > 1:
              self._plot_result_safe(app)
      finally:
          app.close()
  

• Submit (HPC / SLURM mode)

  Inherited from Mintpy_SBAS_Base_Analyzer. Submit full MintPy run as a single sbatch job.

  analyzer.submit_hpc()
  

• Clean up

  Remove intermediate processing files generated during the time-series process

  analyzer.cleanup()
  

  Raises:

  Type	Description
  Exception	Propagates any unexpected errors raised during removal.

The ISCE2_Mintpy_SBAS analyzer extends Mintpy_SBAS_Base_Analyzer and is preconfigured for ISCE2 stackSentinel outputs. prep_data() auto-discovers interferograms and geometry from the isce/ directory and writes the MintPy config to mintpy/.mintpy.cfg. All MintPy outputs are written to workdir/mintpy/.

Source code in src/insarhub/analyzer/isce2_mintpy_s1_sbas.py

class ISCE2_Mintpy_SBAS(Mintpy_SBAS_Base_Analyzer):
    """SBAS time-series analysis of ISCE2 stackSentinel outputs using MintPy.

    Usage::

        from insarhub import Analyzer

        az = Analyzer.create('ISCE2_Mintpy_SBAS', workdir='/data/p64_f468')
        az.prep_data()   # auto-wires MintPy paths, writes mintpy/.mintpy.cfg
        az.run()         # writes all output to workdir/mintpy/
    """

    name                 = "ISCE2_Mintpy_SBAS"
    aliases              = ("ISCE_SBAS", "ISCE2_TS", "ISCE2_Mintpy_TS")   # legacy names
    description          = "SBAS time-series analysis of ISCE2 stackSentinel outputs using MintPy."
    compatible_processor = "ISCE2_S1"
    default_config       = ISCE2_Mintpy_SBAS_Config
    # own output dir (workdir/isce_mintpy/) -- separate from other MintPy
    # analyzers on the same workdir; layout via MintPyPaths (config/paths.py)
    MINTPY_SUBDIR        = "isce_mintpy"

    def __init__(self, config: ISCE2_Mintpy_SBAS_Config | None = None):
        super().__init__(config)
        self._isce_paths = ISCEPaths(self.workdir)
        self.isce_dir    = self._isce_paths.isce_dir

    # ── Public entry points ───────────────────────────────────────────────────

    def prep_data(self) -> None:
        """Auto-discover stackSentinel outputs and write the MintPy config."""
        if self.config.container:
            return self._run_via_container(["prep_data"])

        if not self.isce_dir.exists():
            raise FileNotFoundError(
                f"ISCE processing directory not found: {self.isce_dir}. "
                "Run ISCE2_S1 and wait for all steps to complete."
            )
        ifg_dir = self.isce_dir / "merged" / "interferograms"
        pairs = sorted(d for d in ifg_dir.iterdir() if d.is_dir()) if ifg_dir.exists() else []
        if not pairs:
            raise FileNotFoundError(
                f"No interferogram directories in {ifg_dir}. "
                "ISCE2_S1 processing must reach the interferogram stage first."
            )
        print(f"{Fore.CYAN}Found {len(pairs)} interferogram pair(s). "
              f"Configuring MintPy load paths…{Fore.RESET}")
        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        self._set_load_parameters()
        super().prep_data()   # writes self.cfg_path

    def run(self, steps=None):
        """Run MintPy, writing all output to workdir/mintpy/."""
        if self.config.container:
            return self._run_via_container(steps)

        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        if self.config.troposphericDelay_method == "pyaps" and (steps is None or "correct_troposphere" in steps):
            self._cds_authorize()
        run_steps = steps or [
            "load_data", "modify_network", "reference_point", "quick_overview",
            "correct_unwrap_error", "invert_network", "correct_LOD", "correct_SET",
            "correct_ionosphere", "correct_troposphere", "deramp", "correct_topography",
            "residual_RMS", "reference_date", "velocity", "geocode", "google_earth", "hdfeos5",
        ]
        from colorama import Style
        print(f"{Style.BRIGHT}{Fore.MAGENTA}Running MintPy Analysis…{Fore.RESET}")
        from mintpy.smallbaselineApp import TimeSeriesAnalysis
        app = TimeSeriesAnalysis(self.cfg_path.as_posix(), str(self.mintpy_dir))
        try:
            app.open()
            app.run(steps=run_steps)
            if 'geocode' in run_steps:
                self._geocode_diagnostic_files(self.mintpy_dir)
            # Mirrors mintpy.smallbaselineApp's own CLI wrapper
            # (run_smallbaselineApp()), which calls these two after run() --
            # plot_result() is what actually populates mintpy_dir/pic/, and
            # close() is what restores the process's working directory after
            # open() changed into mintpy_dir (skipping it would leave a
            # long-running server process permanently cd'd into the last
            # analyzed folder).
            if app.template.get('mintpy.plot') and len(run_steps) > 1:
                self._plot_result_safe(app)
        finally:
            app.close()

    def cleanup(self) -> None:
        """Remove large ISCE2 intermediate directories and input data no longer needed
        after MintPy has loaded all data into HDF5.

        Removes under ``isce/``:
          - ``coarse_interferograms/``
          - ``ESD/``
          - ``coreg_secondarys/``
          - ``interferograms/``

        Removes at workdir level:
          - ``slc/``
          - ``dem/``
        """
        if self.config.debug:
            print(f"{Fore.YELLOW}Debug mode enabled. Skipping cleanup.{Fore.RESET}")
            return

        isce_subdirs = [
            self.isce_dir / "coarse_interferograms",
            self.isce_dir / "ESD",
            self.isce_dir / "coreg_secondarys",
            self.isce_dir / "merged" / "interferograms",
        ]
        workdir_subdirs = [
            self._isce_paths.slc_dir,
            self._isce_paths.dem_dir,
        ]

        print(f"{Fore.CYAN}Cleaning up ISCE2 intermediate directories…{Fore.RESET}")
        for folder in isce_subdirs + workdir_subdirs:
            if folder.exists() and folder.is_dir():
                try:
                    shutil.rmtree(folder)
                    print(f"  Removed: {folder.relative_to(self.workdir)}")
                except Exception as e:
                    print(f"{Fore.RED}  Failed to remove {folder}: {e}{Fore.RESET}")
            else:
                print(f"  Skipped (not found): {folder.relative_to(self.workdir)}")
        print(f"{Fore.GREEN}Cleanup complete.{Fore.RESET}")

    # ── Path discovery ────────────────────────────────────────────────────────

    def _set_load_parameters(self) -> None:
        """Wire all MintPy load_* fields from the stackSentinel output layout.

        Handles both geocoded (.geo suffix) and radar-coordinate (no suffix)
        variants, and both merged/geometry/ and merged/geom_reference/ layouts.
        """
        isce   = self.isce_dir
        merged = isce / "merged"

        # ── interferogram files: prefer geocoded, fall back to radar-coordinate ──
        ifg_base = merged / "interferograms"
        sample_pair = next((d for d in ifg_base.iterdir() if d.is_dir()), None) if ifg_base.exists() else None

        def _ifg_file(stem: str) -> str:
            if sample_pair and (sample_pair / f"{stem}.geo").exists():
                return str(ifg_base / "*" / f"{stem}.geo")
            return str(ifg_base / "*" / stem)

        self.config.load_unwFile      = _ifg_file("filt_fine.unw")
        self.config.load_corFile      = _ifg_file("filt_fine.cor")
        self.config.load_connCompFile = _ifg_file("filt_fine.unw.conncomp")

        # ── geometry: prefer merged/geometry/, fall back to merged/geom_reference/ ──
        geo = merged / "geometry" if (merged / "geometry").exists() else merged / "geom_reference"

        def _geo(name_geo: str, name_rdr: str) -> str:
            f = geo / name_geo
            return str(f) if f.exists() else str(geo / name_rdr)

        self.config.load_demFile      = _geo("hgt.rdr.geo",       "hgt.rdr")
        self.config.load_incAngleFile = _geo("incLocal.rdr.geo",   "incLocal.rdr")
        self.config.load_lookupYFile  = _geo("lat.rdr.geo",        "lat.rdr")
        self.config.load_lookupXFile  = _geo("lon.rdr.geo",        "lon.rdr")

        # azimuth angle: dedicated file or fall back to los.rdr (band 2)
        az_file = geo / "azimuthAngle.rdr.geo"
        if not az_file.exists():
            az_file = geo / "azimuthAngle.rdr"
        if not az_file.exists():
            az_file = geo / "los.rdr"   # band 2 = azimuth angle
        self.config.load_azAngleFile = str(az_file)

        # shadow/layover mask
        for sn in ("shadowMask.rdr.geo", "shadowMask.rdr"):
            sf = geo / sn
            if sf.exists():
                self.config.load_shadowMaskFile = str(sf)
                break

        # water mask (pre-existing only)
        for wn in ("waterMask.geo", "waterMask.rdr.geo", "waterMask.rdr"):
            wf = geo / wn
            if wf.exists():
                self.config.load_waterMaskFile = str(wf)
                break
        else:
            self.config.load_waterMaskFile = "no"

        self.config.load_baselineDir = str(isce / "baselines")
        self.config.load_metaFile    = self._find_meta_file()

        print(f"{Fore.GREEN}  unwFile      : {self.config.load_unwFile}")
        print(f"  corFile      : {self.config.load_corFile}")
        print(f"  demFile      : {self.config.load_demFile}")
        print(f"  geometry dir : {geo}")
        print(f"  metaFile     : {self.config.load_metaFile}")
        print(f"  baselineDir  : {self.config.load_baselineDir}{Fore.RESET}")

    def _find_meta_file(self) -> str:
        """Locate the reference IW*.xml metadata file in the stackSentinel tree."""
        isce = self.isce_dir

        # isce/reference/IW*.xml  (most common stackSentinel layout)
        ref_dir = isce / "reference"
        if ref_dir.exists():
            xmls = sorted(ref_dir.glob("IW*.xml"))
            if xmls:
                return str(xmls[0])
            # isce/reference/{date}/IW*.xml
            for sub in sorted(ref_dir.iterdir()):
                if sub.is_dir():
                    xmls = sorted(sub.glob("IW*.xml"))
                    if xmls:
                        return str(xmls[0])

        # isce/merged/SLC/{date}/*.xml  (stackSentinel merged SLC layout)
        slc_merged = isce / "merged" / "SLC"
        if slc_merged.exists():
            for date_dir in sorted(slc_merged.iterdir()):
                if date_dir.is_dir():
                    xmls = sorted(date_dir.glob("*.slc.full.xml"))
                    if not xmls:
                        xmls = sorted(date_dir.glob("*.xml"))
                    if xmls:
                        return str(xmls[0])

        # fall back: pass the reference directory and let MintPy scan it
        return str(ref_dir)

Usage

• Create Analyzer

  from insarhub import Analyzer
  
  analyzer = Analyzer.create('ISCE2_Mintpy_SBAS', workdir='/your/work/dir')
  

  OR with explicit config:

  from insarhub.config.defaultconfig import ISCE2_Mintpy_SBAS_Config
  
  cfg = ISCE2_Mintpy_SBAS_Config(workdir='/your/work/dir')
  analyzer = Analyzer.create('ISCE2_Mintpy_SBAS', config=cfg)
  

• Prepare data

  Auto-discover ISCE2 outputs and write mintpy/.mintpy.cfg.

  analyzer.prep_data()
  

• Run

  Run MintPy SBAS time-series analysis. All output written to workdir/mintpy/.

  analyzer.run()
  

• Submit (HPC / SLURM mode)

  Inherited from Mintpy_SBAS_Base_Analyzer. Submit full MintPy run as a single sbatch job.

  analyzer.submit_hpc()
  

• Clean up

  Remove large ISCE2 intermediate directories and input data no longer needed after load_data. Removes isce/coarse_interferograms/, isce/ESD/, isce/coreg_secondarys/, isce/interferograms/, slc/, and dem/.

  analyzer.cleanup()
  

Runs MintPy SBAS on the stack from the GMTSAR_S1 processor, handing GMTSAR's geocoded *_ll.grd products and baseline_table.dat to MintPy's prep_gmtsar.py loader. The MintPy analogue of ISCE2_Mintpy_SBAS. Output goes to workdir/gmtsar_mintpy/, kept separate so it never collides with a Hyp3 or ISCE MintPy run in the same workdir.

Source code in src/insarhub/analyzer/gmtsar_mintpy_s1_sbas.py

class GMTSAR_Mintpy_SBAS(Mintpy_SBAS_Base_Analyzer):
    name                 = "GMTSAR_Mintpy_SBAS"
    aliases              = ("GMTSAR_MINTPY_SBAS", "GMTSAR_MINTPY_TS", "GMTSAR_Mintpy_TS")   # legacy names
    description          = "SBAS time-series of a GMTSAR stack_mode stack using MintPy (prep_gmtsar.py)."
    compatible_processor = "GMTSAR_S1"
    default_config       = GMTSAR_Mintpy_SBAS_Config
    # own output dir (workdir/gmtsar_mintpy/) -- separate from Hyp3_Mintpy_SBAS/
    # ISCE2_Mintpy_SBAS runs on the same workdir; layout via MintPyPaths
    MINTPY_SUBDIR        = "gmtsar_mintpy"

    def __init__(self, config: GMTSAR_Mintpy_SBAS_Config | None = None):
        super().__init__(config)
        self._gmtsar_paths = GMTSARPaths(Path(self.workdir))
        #: subswath _collect_date_prms pinned the baselines to (p2p only)
        self._baseline_swath: int | None = None

    @property
    def stack_dir(self) -> Path:
        """Where GMTSAR_S1 stack_mode wrote its output."""
        return self._gmtsar_paths.case_dir

    #: The acquisition start time in a .SAFE name: S1A_..._20210108T133459_...
    _SAFE_DATE_RE = re.compile(r"(\d{8})T\d{6}")

    @staticmethod
    def _to_gmtsar_julian(yyyymmdd: str) -> str:
        """``20210108`` -> ``2021007``, GMTSAR's pair-directory naming.

        The day field is the offset from Jan 1, not the 1-based day of year:
        8 Jan is 007, matching what preproc_batch_tops writes and what
        MintPy's ptime.yyyyddd2yyyymmdd() reads back.
        """
        from datetime import date
        d = date(int(yyyymmdd[:4]), int(yyyymmdd[4:6]), int(yyyymmdd[6:8]))
        return f"{d.year}{(d - date(d.year, 1, 1)).days:03d}"

    def _p2p_product_dir(self) -> Path | None:
        """Stage p2p output into the ``<dir>/<pair>/unwrap_ll.grd`` shape both
        MintPy and _consistent_intf_dir expect. None when this isn't p2p output.

        The two modes put the merged product at different depths::

            stack_mode   merge/2021007_2021019/unwrap_ll.grd
            p2p          gmtsar/<ref>.SAFE_<sec>.SAFE/merge/unwrap_ll.grd

        because p2p_S1_TOPS_Frame always writes merge/ into its own cwd, so
        GMTSAR_S1 gives every pair its own case dir to stop pair 2 overwriting
        pair 1. That extra level means the ``<intf>/*/unwrap_ll.grd`` glob
        matches nothing at gmtsar/ (too shallow) and exactly one pair at a case
        dir. Symlinking sidesteps it without copying multi-GB grids.

        The staged directories MUST keep GMTSAR's Julian ``yyyyddd_yyyyddd``
        naming. prep_gmtsar derives the pair from the directory name and
        converts it itself::

            date1, date2 = os.path.basename(ifg_dir).split('_')
            date1 = ptime.yyyyddd2yyyymmdd(date1)

        so a calendar-named ``20210108_20210120`` is read as year 2021 day 120
        -> 20210501, which is not in the baseline table: ``KeyError: '20210501'``.
        (Tried it -- naming them by real dates looks tidier and would also fix
        MintPy's skip_files_with_inconsistent_size(), which matches ``yymmdd``
        against the path and so never matches a Julian directory. But that
        function is a nicety and prep_gmtsar is mandatory, so Julian wins.)
        """
        import shutil

        found = sorted(self.stack_dir.glob("*/merge/unwrap_ll.grd"))
        if not found:
            return None
        dest = self.mintpy_dir / "p2p_pairs"
        if dest.exists():
            shutil.rmtree(dest)
        dest.mkdir(parents=True)

        staged = 0
        for unw in found:
            case = unw.parent.parent
            # 4 timestamps in "<ref>.SAFE_<sec>.SAFE" (start/stop each); the
            # two starts are the pair's dates.
            dates = self._SAFE_DATE_RE.findall(case.name)
            if len(dates) < 3:
                logger.warning("p2p: no date pair in case name %r; skipped", case.name)
                continue
            out = dest / f"{self._to_gmtsar_julian(dates[0])}_" \
                         f"{self._to_gmtsar_julian(dates[2])}"
            out.mkdir(exist_ok=True)
            ok = True
            for name in ("unwrap_ll.grd", "corr_ll.grd"):
                src = unw.parent / name
                if not src.exists():
                    logger.warning("p2p: %s has no %s; pair skipped", case.name, name)
                    ok = False
                    break
                (out / name).symlink_to(src)
            if ok:
                staged += 1
            else:
                shutil.rmtree(out, ignore_errors=True)

        if not staged:
            return None
        print(f"{Fore.CYAN}p2p layout: staged {staged} pair(s) as "
              f"<date>_<date>/ under {dest}{Fore.RESET}")
        return dest

    def _product_dir(self) -> Path:
        """The per-pair product dir for either mode (p2p staged, else native)."""
        if getattr(self, "_product_dir_cache", None) is None:
            self._product_dir_cache = (self._p2p_product_dir()
                                       or self._gmtsar_paths.product_dir())
        return self._product_dir_cache

    def prep_data(self) -> None:
        """Auto-discover the GMTSAR stack outputs and write the MintPy config."""
        if self.config.container:
            return self._run_via_container(["prep_data"])

        stack = self.stack_dir
        # preproc_batch_tops writes baseline_table.dat into raw/, not the root.
        baseline = self._gmtsar_paths.baseline_table_auto
        intf_dir = self._product_dir()
        if not baseline.exists():
            # p2p mode never produces one: baseline_table.dat is written by
            # stack_mode's preproc_batch_tops, which is the only stage that
            # sees every date at once. Build it from the per-date PRMs the
            # p2p pairs already left behind rather than requiring stack_mode,
            # which the GMTSAR developers advise against for interferogram
            # formation. Measured on pair 2021127_2021139, processed both
            # ways on p100_f466: 24 azimuth seam rows (>5 sigma) for
            # stack_mode vs 6 for p2p, and the 6 are the frame-edge pair
            # both modes share -- so 18 interior burst seams vs 0. Coherence
            # was equivalent (0.628 vs 0.634), so this is alignment, not SNR.
            built = self._build_baseline_table(baseline)
            if not built:
                raise FileNotFoundError(
                    f"{baseline} not found and could not be built from the "
                    f"per-date PRMs. MintPy's prep_gmtsar needs one row per "
                    f"date (file_ID, yyyyddd.fraction, day_cnt, b_para, "
                    f"b_perp). Either run GMTSAR_S1 with stack_mode=True, or "
                    f"check that the pairs left per-date PRMs behind "
                    f"(S1_<date>_ALL_F<n>.PRM in stack_mode, "
                    f"S1_<date>_<time>_F<n>.PRM in p2p)."
                )
        pairs = sorted(d for d in intf_dir.iterdir() if d.is_dir()) if intf_dir.exists() else []
        if not pairs:
            raise FileNotFoundError(
                f"No interferogram directories in {intf_dir}. GMTSAR_S1 "
                "stack_mode must reach the intf stage first."
            )
        print(f"{Fore.CYAN}Found {len(pairs)} interferogram pair(s). "
              f"Configuring MintPy (prep_gmtsar) load paths…{Fore.RESET}")
        self.mintpy_dir.mkdir(parents=True, exist_ok=True)
        self._set_load_parameters()
        super().prep_data()   # writes self.cfg_path
        self._ensure_prep_gmtsar_inputs()

    # Two different PRM namings, because the two GMTSAR entry points name
    # them differently:
    #   stack_mode  preproc_batch_tops -> S1_20210108_ALL_F1.PRM
    #   p2p (Frame) p2p_S1_TOPS_Frame  -> S1_20210108_133500_F1.PRM
    # Matching only the first is why MintPy SBAS could never run on p2p
    # output: the glob returned nothing, _build_baseline_table bailed at
    # "need at least 2", and no baseline_table.dat was ever written.
    _PRM_DATE_RE = re.compile(r"_(\d{8})_(?:ALL|\d{6})_F(\d)")

    def _collect_date_prms(self) -> dict[str, Path]:
        """``{YYYYMMDD: PRM}`` -- one PRM per acquisition date.

        p2p processes each pair in its own directory, so the same date's PRM
        appears once per pair it takes part in. Any copy will do: baselines are
        computed from the ORBIT state vectors, which belong to the date itself,
        not to whichever pair happened to produce that copy. Alignment changes
        rshift/ashift, not the orbit -- so this stays correct even though p2p
        aligns each pair to its own master.
        """
        by_swath: dict[str, dict[str, Path]] = {}
        for prm in sorted(self.stack_dir.rglob("*.PRM")):
            m = self._PRM_DATE_RE.search(prm.name)
            if not m:
                continue      # e.g. topo/master.PRM -- no date in the name
            date, swath = m.group(1), m.group(2)
            slot = by_swath.setdefault(swath, {})
            # Prefer the raw/ copy: it carries the orbit straight off the EOF,
            # and its .LED sits beside it (baseline_table.csh resolves led_file
            # relative to the PRM, and we run it with cwd=the PRM's parent).
            if date not in slot or (prm.parent.name == "raw"
                                    and slot[date].parent.name != "raw"):
                slot[date] = prm
        if not by_swath:
            return {}
        # Pin to ONE subswath. b_perp depends on look geometry, which differs
        # slightly between IWs, so taking F1 for one date and F2 for another
        # would fold that difference into the baselines. Best-covered wins.
        swath = max(by_swath, key=lambda s: len(by_swath[s]))
        # Recorded so _set_load_parameters can pull the metadata PRM from the
        # same subswath -- metadata and baselines should describe one geometry.
        self._baseline_swath = int(swath)
        return by_swath[swath]

    def _build_baseline_table(self, dest: Path) -> bool:
        """Write ``baseline_table.dat`` from per-date PRMs via GMTSAR's own
        ``baseline_table.csh``.

        The reference is the EARLIEST date, matching what preproc_batch_tops
        uses, so a table built here is interchangeable with a stack-mode one.
        Note the reference only sets the zero point of the perpendicular
        baselines -- it is a geometry calculation, and nothing is resampled --
        which is why this works for p2p output that has no common alignment
        master at all. MintPy reads only the date and b_perp columns.

        Every PRM is staged into one directory alongside its ``.LED`` first.
        ``baseline_table.csh`` resolves each PRM's ``led_file`` (a bare
        filename) relative to the WORKING directory, and p2p keeps each date's
        PRM in its own case dir -- so calling it across directories silently
        finds no orbit for the secondary and prints 3 columns instead of 5.
        MintPy then reads the table with ``usecols=(1, 4)`` and dies with
        ``invalid column index 4 ... with 3 columns``, having already been told
        by load_data that "prep_gmtsar.py failed, assuming its result exists"
        -- so the real cause surfaces much later as ``KeyError: 'DATE12'``.
        """
        import shutil, subprocess, tempfile

        prms = self._collect_date_prms()
        if len(prms) < 2:
            logger.warning("GMTSAR_Mintpy_SBAS: found %d date PRM(s) under %s; "
                           "need at least 2 to build a baseline table",
                           len(prms), self.stack_dir)
            return False

        tmpdir = Path(tempfile.mkdtemp(prefix="baseline_", dir=str(self.mintpy_dir)))
        staged: dict[str, Path] = {}
        for date, src in prms.items():
            led = src.with_suffix(".LED")
            if not led.exists():
                logger.warning("GMTSAR_Mintpy_SBAS: %s has no .LED beside it; "
                               "baselines for %s would be wrong", src.name, date)
                shutil.rmtree(tmpdir, ignore_errors=True)
                return False
            shutil.copy(src, tmpdir / src.name)
            shutil.copy(led, tmpdir / led.name)
            staged[date] = tmpdir / src.name
        prms = staged

        ref_date = min(prms)
        ref_prm = prms[ref_date]
        print(f"{Fore.CYAN}Building baseline_table.dat from {len(prms)} date "
              f"PRM(s), reference {ref_date}…{Fore.RESET}")

        rows: list[str] = []
        try:
            for d in sorted(prms):
                p = subprocess.run(
                    ["baseline_table.csh", ref_prm.name, prms[d].name],
                    text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                    cwd=str(tmpdir))
                line = (p.stdout or "").strip()
                if p.returncode != 0 or not line:
                    logger.error("GMTSAR_Mintpy_SBAS: baseline_table.csh failed "
                                 "for %s: %s", d, (p.stderr or p.stdout or "")[-300:])
                    return False
                row = line.splitlines()[0]
                # 5 columns: file_ID, yyyyddd.frac, day_cnt, b_para, b_perp.
                # Fewer means the orbit was not resolved and the baselines are
                # simply absent -- catch it here rather than letting MintPy hit
                # "invalid column index 4" three steps downstream.
                if len(row.split()) < 5:
                    logger.error(
                        "GMTSAR_Mintpy_SBAS: baseline_table.csh returned %d "
                        "columns for %s (need 5). The .LED beside %s is "
                        "probably unreadable:\n  %s",
                        len(row.split()), d, prms[d].name, row[:160])
                    return False
                rows.append(row)

            dest.parent.mkdir(parents=True, exist_ok=True)
            dest.write_text("\n".join(rows) + "\n")
            print(f"{Fore.GREEN}  wrote {dest} ({len(rows)} dates, "
                  f"{len(rows[0].split())} columns){Fore.RESET}")
            return True
        finally:
            shutil.rmtree(tmpdir, ignore_errors=True)

    def _ensure_prep_gmtsar_inputs(self) -> None:
        """Two extra inputs MintPy's prep_gmtsar.py requires that GMTSAR
        stack_mode doesn't produce (found via a real failed run):

        1. config.<SAT>.txt next to the metaFile PRM -- prep_gmtsar reads
           filter_wavelength from it to compute ALOOKS/RLOOKS. GMTSAR_S1
           writes batch_tops.config instead, so mirror the value into the
           file prep_gmtsar globs for.
        2. a bare HEADING key in the template -- not derivable from GMTSAR's
           PRM (orbdir=D/A only, no angle). Use MintPy's own canonical
           values for Sentinel-1 IW (utils0.py): -168 deg descending,
           -12 deg ascending, picked from the PRM's orbdir.
        """
        from pathlib import Path
        meta = Path(self.config.load_metaFile)
        raw = meta.parent

        if not list(raw.glob("config.*.txt")):
            fw = "200"
            btc = self._gmtsar_paths.batch_config
            srcs = [btc] if btc.exists() else []
            # p2p writes config.py per case instead of batch_tops.config, so
            # without this the hardcoded 200 was used regardless of the real
            # setting -- and MintPy derives ALOOKS/RLOOKS from this value. A
            # stack processed at 320 was being described to MintPy as 200.
            srcs += sorted(self.stack_dir.glob("*/config.py"))[:1]
            for src in srcs:
                for l in src.read_text().splitlines():
                    if l.strip().startswith("filter_wavelength"):
                        fw = l.split("=")[1].strip()
                        break
                else:
                    continue
                break
            (raw / "config.S1_TOPS.txt").write_text(f"filter_wavelength = {fw}\n")
            print(f"  wrote {raw / 'config.S1_TOPS.txt'} (filter_wavelength = {fw})")

        cfg_text = self.cfg_path.read_text() if self.cfg_path.exists() else ""
        if "HEADING" not in cfg_text:
            orbdir = "D"
            if meta.exists():
                for l in meta.read_text().splitlines():
                    if l.strip().startswith("orbdir"):
                        orbdir = l.split("=")[1].strip().upper()[:1]
                        break
            heading = -168.0 if orbdir == "D" else -12.0
            with open(self.cfg_path, "a") as f:
                f.write(f"HEADING = {heading}\n")
            print(f"  appended HEADING = {heading} (orbdir={orbdir}) to {self.cfg_path}")

    def _gdal_env(self) -> dict:
        """Bare subprocess env lacks the conda activation vars gdal/gmt need
        to resolve EPSG codes (real failure: "proj_create_from_database:
        Open of .../share/proj failed")."""
        import os as _os, sys as _sys
        env = _os.environ.copy()
        envroot = None
        if getattr(self.config, "gmtsar_env_bin", None):
            envroot = str(Path(self.config.gmtsar_env_bin).parent)
            env["PATH"] = f"{self.config.gmtsar_env_bin}:" + env.get("PATH", "")
        # gmtsar_env_bin lives on the PROCESSOR config; the analyzer's own
        # config usually has none, which left PROJ_DATA unset and gdal_translate
        # failing with the very error this method exists to prevent. The
        # running interpreter is already inside the right env, so use its
        # prefix. PROJ_LIB as well as PROJ_DATA -- PROJ reads the old name.
        if envroot is None or not Path(f"{envroot}/share/proj").is_dir():
            envroot = _sys.prefix
        if Path(f"{envroot}/share/proj").is_dir():
            env.setdefault("PROJ_DATA", f"{envroot}/share/proj")
            env.setdefault("PROJ_LIB", f"{envroot}/share/proj")
        if Path(f"{envroot}/share/gdal").is_dir():
            env.setdefault("GDAL_DATA", f"{envroot}/share/gdal")
        return env

    @staticmethod
    def _grd_shape(grd: Path) -> tuple[int, int] | None:
        try:
            from netCDF4 import Dataset
            with Dataset(grd) as ds:
                for name in ("z", "Band1"):
                    if name in ds.variables:
                        return tuple(ds.variables[name].shape)
        except Exception:
            return None
        return None

    def _grd_region(self, grd: Path) -> tuple[float, float, float, float] | None:
        """(x_min, x_max, y_min, y_max) of a .grd."""
        try:
            from netCDF4 import Dataset
            with Dataset(grd) as ds:
                x = ds.variables.get("lon", ds.variables.get("x"))
                y = ds.variables.get("lat", ds.variables.get("y"))
                if x is None or y is None:
                    return None
                return (float(x[0]), float(x[-1]), float(y[0]), float(y[-1]))
        except Exception:
            return None

    def _grd_increments(self, grd: Path) -> tuple[float, float] | None:
        """(x_inc, y_inc) of a .grd."""
        try:
            from netCDF4 import Dataset
            with Dataset(grd) as ds:
                x = ds.variables.get("lon", ds.variables.get("x"))
                y = ds.variables.get("lat", ds.variables.get("y"))
                if x is None or y is None or len(x) < 2 or len(y) < 2:
                    return None
                return (abs(float(x[1]) - float(x[0])), abs(float(y[1]) - float(y[0])))
        except Exception:
            return None

    def _consistent_intf_dir(self, intf_all: Path) -> Path:
        """Return a directory whose */unwrap_ll.grd all share ONE common grid,
        clipping every pair to the common overlap when they don't already.

        GMTSAR geocodes each pair to its own valid-data EXTENT but the SAME grid
        (identical pixel size + registration -- origins differ only by whole
        pixels; measured here: 540x540 vs 540x550, y-origins exactly 10 pixels
        apart). MintPy's skip_files_with_inconsistent_size() detects the size
        mismatch but can't drop it (it matches `yymmdd`, and GMTSAR names dirs by
        Julian date), so the file loads anyway -> "can't broadcast".

        Because the pairs share one underlying grid, the reconciliation is an
        exact integer-pixel CLIP onto the intersection extent -- no resampling
        (this is what HyP3's prep does: clip products to their common overlap).
        The bogus header nodata (a plain number, the Julian ref date e.g.
        2024025 -- NOT NaN, which `gmt grdcut` mishandles into 50-70% spurious
        NaN that zeroes the inversion) is masked to real NaN in NumPy; the clip
        + netCDF write go through gdal ReadAsArray/Translate. A nearest-neighbour
        warp is used only as a fallback if a pair is ever sub-pixel-misaligned.
        (We avoid gdal.Warp with srcNodata+dstNodata: GDAL 3.6.3 segfaults on
        that exact combination.)
        """
        import shutil
        import numpy as np
        from osgeo import gdal
        gdal.UseExceptions()

        pairs = sorted(d for d in intf_all.iterdir() if d.is_dir())
        geo: dict[Path, tuple] = {}     # dir -> (geotransform, W, H)
        for d in pairs:
            u = d / "unwrap_ll.grd"
            if not u.exists():
                continue
            ds = gdal.Open(str(u))
            geo[d] = (ds.GetGeoTransform(), ds.RasterXSize, ds.RasterYSize)
            ds = None
        if not geo:
            return intf_all
        if len({(W, H) for _, W, H in geo.values()}) == 1:
            return intf_all             # already one grid -> raw grids are fine

        for d, (_, W, H) in geo.items():
            print(f"  pair {d.name}: {W}x{H}")

        # Common overlap = the INTERSECTION extent. GMTSAR geocodes every pair
        # to the SAME posting and grid registration (identical pixel size, and
        # origins that differ only by whole pixels), so the pairs share one
        # underlying grid and merely cover different extents. Reconcile them by
        # an exact integer-pixel CLIP onto the common overlap -- no resampling
        # (this is what HyP3's prep does: clip products to their common overlap).
        # Only if a pair turns out sub-pixel-misaligned do we fall back to a
        # nearest-neighbour warp.
        gt0 = next(iter(geo.values()))[0]
        px, py = gt0[1], gt0[5]                       # x_step (+), y_step (-)
        minx = max(gt[0] for gt, _, _ in geo.values())
        maxy = min(gt[3] for gt, _, _ in geo.values())
        maxx = min(gt[0] + w * gt[1] for gt, w, _ in geo.values())
        miny = max(gt[3] + h * gt[5] for gt, _, h in geo.values())
        W = int(round((maxx - minx) / abs(px)))
        H = int(round((maxy - miny) / abs(py)))
        aligned = all(
            abs((minx - gt[0]) / px - round((minx - gt[0]) / px)) < 1e-3
            and abs((maxy - gt[3]) / py - round((maxy - gt[3]) / py)) < 1e-3
            for gt, _, _ in geo.values())
        new_gt = (minx, px, 0.0, maxy, 0.0, py)
        print(f"  reconciling {len(geo)} pairs onto a common {W}x{H} grid "
              f"({'exact integer-pixel clip' if aligned else 'nearest-neighbour resample'})")

        links = self.clip_dir
        if links.exists():
            shutil.rmtree(links)
        links.mkdir(parents=True)

        kept = 0
        for d in geo:
            out = links / d.name
            out.mkdir()
            ok = True
            for name in ("unwrap_ll.grd", "corr_ll.grd"):
                src = d / name
                if not src.exists():
                    ok = False
                    break
                try:
                    # Invalid pixels carry a bogus NODATA value that is a plain
                    # number (the pair's Julian ref date, e.g. 2024001) declared
                    # in the header -- NOT NaN. Mask it to real NaN so MintPy/sbas
                    # treat it as no-data instead of ~2e6 "phase". (Do NOT use
                    # gdal.Warp with srcNodata+dstNodata: GDAL 3.6.3 here
                    # SEGFAULTs on that combo -- so we clip in NumPy and write
                    # netCDF via Translate, whose write path is stable.)
                    sd = gdal.Open(str(src))
                    nd = sd.GetRasterBand(1).GetNoDataValue()
                    gt = sd.GetGeoTransform()
                    if aligned:
                        xoff = int(round((minx - gt[0]) / px))
                        yoff = int(round((maxy - gt[3]) / py))
                        arr = sd.GetRasterBand(1).ReadAsArray(xoff, yoff, W, H)
                        sd = None
                        arr = arr.astype("float32")
                        if nd is not None:
                            arr = np.where(arr == nd, np.nan, arr)
                    else:
                        a = sd.GetRasterBand(1).ReadAsArray().astype("float32")
                        if nd is not None:
                            a = np.where(a == nd, np.nan, a)
                        src_mem = gdal.GetDriverByName("MEM").Create(
                            "", sd.RasterXSize, sd.RasterYSize, 1, gdal.GDT_Float32)
                        src_mem.SetGeoTransform(gt)
                        src_mem.GetRasterBand(1).WriteArray(a)
                        sd = None
                        w = gdal.Warp("", src_mem, format="MEM",
                                      outputBounds=(minx, miny, maxx, maxy),
                                      width=W, height=H, resampleAlg="near")
                        src_mem = None
                        arr = w.GetRasterBand(1).ReadAsArray()
                        w = None
                    m = gdal.GetDriverByName("MEM").Create("", W, H, 1, gdal.GDT_Float32)
                    m.SetGeoTransform(new_gt)
                    m.GetRasterBand(1).WriteArray(arr)
                    m.GetRasterBand(1).SetNoDataValue(float("nan"))
                    gdal.Translate(str(out / name), m, format="netCDF")
                    m = None
                except Exception as exc:                 # noqa: BLE001
                    logger.warning("clip failed for %s/%s: %s", d.name, name, exc)
                    ok = False
                    break
            if ok:
                kept += 1
            else:
                shutil.rmtree(out, ignore_errors=True)

        print(f"{Fore.GREEN}  reconciled {kept}/{len(geo)} pairs -> {links} "
              f"(uniform {W}x{H} master grid){Fore.RESET}")
        return links

    def _meta_raw_dir(self) -> Path:
        """The raw/ of the subswath whose output actually populated merge/.

        _gmtsar_paths.meta_raw_dir blindly returns the FIRST F<N>/raw, which is
        wrong for an AOI-narrowed stack: it processes a single subswath that may
        not be F1 (e.g. an AOI only in F2), while F1/raw still holds stale
        aligned PRMs + baseline_table.dat from an earlier full-frame run. The
        single-subswath merge output is a symlink into F<sw>/intf_all/, so
        resolve one merge/<pair>/unwrap_ll.grd back to its real F<sw>/raw. Falls
        back to meta_raw_dir for a true multi-subswath merge (whose merged
        product is a real file, not a symlink into a subswath)."""
        default = self._gmtsar_paths.meta_raw_dir
        for u in sorted(self._product_dir().glob("*/unwrap_ll.grd")):
            try:
                real = u.resolve()
            except OSError:
                continue
            if real == u:
                continue                       # not a symlink -> real merge
            for parent in real.parents:
                if parent.name == "intf_all":
                    raw = parent.parent / "raw"
                    if raw.is_dir() and list(raw.glob("S1_*_ALL_F*.PRM")):
                        return raw
                    break
        return default

    def _set_load_parameters(self) -> None:
        """Wire the mintpy.load.* keys prep_gmtsar.py reads from GMTSAR's
        geocoded stack output. prep_gmtsar globs `<fbase>_ll*.grd` and derives
        LAT/LON_REF + geo-transform from the *_ll.grd files themselves, so the
        essential inputs are the unwrapped/coherence _ll grids, one sample PRM
        (metadata), and baseline_table.dat (per-date baselines)."""
        stack = self.stack_dir
        intf = self._consistent_intf_dir(self._product_dir())

        self.config.load_unwFile     = str(intf / "*" / "unwrap_ll.grd")
        self.config.load_corFile     = str(intf / "*" / "corr_ll.grd")

        # metadata + per-date baselines from the subswath that actually produced
        # merge/ (not a stale sibling F<N>/raw) -- see _meta_raw_dir.
        raw = self._meta_raw_dir()
        self.config.load_baselineDir = str(raw / "baseline_table.dat")
        prm = (next(iter(sorted(raw.glob("S1_*_ALL_F*.PRM"))), None)
               or next(iter(sorted(raw.glob("S1_*.PRM"))), None))
        if prm is None:
            # p2p keeps its PRMs per case, at <case>/F<N>/raw/, so meta_raw_dir
            # -- which assumes the stack_mode layout -- is empty. Falling
            # through to a "<raw>/*.PRM" glob looked harmless but pointed
            # MintPy at a directory holding only the two files this analyzer
            # had just written, and prep_gmtsar needs a real PRM for the radar
            # wavelength and orbit direction.
            #
            # Take the same subswath the baselines came from, so metadata and
            # b_perp describe one geometry rather than two.
            same_swath = sorted(self.stack_dir.glob(
                f"*/F{self._baseline_swath or 1}/raw/S1_*_F?.PRM"))
            prm = next(iter(same_swath),
                       next(iter(sorted(self.stack_dir.glob("*/F?/raw/S1_*_F?.PRM"))), None))
            if prm is not None:
                logger.info("p2p layout: metaFile taken from %s",
                            prm.relative_to(self.stack_dir))
        self.config.load_metaFile = str(prm) if prm else str(raw / "*.PRM")

        # DEM: two things must be fixed before MintPy can use GMTSAR's dem.grd
        #  1. it carries no projection -> EPSG=None -> HDF5 attr write crash
        #     ("Object dtype dtype('O') has no native HDF5 equivalent")
        #  2. it spans the whole SLC footprint at 3-arcsec, while the stack is
        #     a sub-region at 2-arcsec -> geometryGeo.h5 (2671,4121) vs
        #     ifgramStack (1830,1870) -> "could not broadcast" in plot_result.
        # So resample it onto the stack's exact grid, then stamp EPSG:4326.
        # (grdsample also reconciles the 0-360 vs -180-180 longitude
        # convention: GMTSAR geocodes to 246..247, the DEM is -114..-111.)
        # dem_grd assumes the stack layout (gmtsar/topo/dem.grd). p2p keeps the
        # shared DEM at the workdir root and symlinks each case at it, so fall
        # back through both. Getting this wrong left demFile="auto", and
        # prep_gmtsar does glob.glob("auto")[0] -> IndexError with no message
        # naming the DEM at all.
        src_dem = next(
            (p for p in (self._gmtsar_paths.dem_grd,
                         Path(self.workdir) / "topo" / "dem.grd",
                         *sorted(self.stack_dir.glob("*/topo/dem.grd"))[:1])
             if p.exists()),
            self._gmtsar_paths.dem_grd)
        sample = next(iter(sorted(intf.glob("*/unwrap_ll.grd"))), None)
        if src_dem.exists() and sample is not None:
            logger.info("DEM for MintPy: %s", src_dem)
            import subprocess as _sp
            env = self._gdal_env()
            reg, inc = self._grd_region(sample), self._grd_increments(sample)
            dem_tif = self.mintpy_dir / "dem_match.tif"
            if reg and inc:
                tmp = self.mintpy_dir / "dem_match.grd"
                _sp.run(["gmt", "grdsample", str(src_dem),
                         f"-R{reg[0]}/{reg[1]}/{reg[2]}/{reg[3]}",
                         f"-I{inc[0]}/{inc[1]}", f"-G{tmp}"],
                        check=True, capture_output=True, env=env)
                _sp.run(["gdal_translate", "-a_srs", "EPSG:4326",
                         str(tmp), str(dem_tif)],
                        check=True, capture_output=True, env=env)
                print(f"  DEM resampled onto the stack grid "
                      f"{self._grd_shape(tmp)} -> {dem_tif.name}")
                self.config.load_demFile = str(dem_tif)

        # GMTSAR's geocoded stack has no lookup-table / incidence / mask
        # rasters -- MintPy derives those from the *_ll.grd geometry itself.
        # They must be EMPTY, not "auto": prep_gmtsar treats any truthy value
        # as a real glob pattern (`glob.glob(template[key])[0]`), so "auto"
        # raises IndexError and the whole prep step silently fails (MintPy
        # swallows it as "Assuming its result exists"), leaving no per-pair
        # .rsc -> later KeyError: 'DATE12'. Found via a real run.
        for key in ("load_lookupYFile", "load_lookupXFile", "load_incAngleFile",
                    "load_azAngleFile", "load_shadowMaskFile", "load_waterMaskFile",
                    "load_connCompFile"):
            if hasattr(self.config, key) and getattr(self.config, key) == "auto":
                setattr(self.config, key, "")

        print(f"{Fore.GREEN}  unwFile     : {self.config.load_unwFile}")
        print(f"  corFile     : {self.config.load_corFile}")
        print(f"  metaFile    : {self.config.load_metaFile}")
        print(f"  baselineDir : {self.config.load_baselineDir}{Fore.RESET}")

Usage

• Create Analyzer

  from insarhub import Analyzer
  
  analyzer = Analyzer.create('GMTSAR_Mintpy_SBAS', workdir='/your/work/dir')
  

  OR with explicit config:

  from insarhub.config.defaultconfig import GMTSAR_Mintpy_SBAS_Config
  
  cfg = GMTSAR_Mintpy_SBAS_Config(workdir='/your/work/dir')
  analyzer = Analyzer.create('GMTSAR_Mintpy_SBAS', config=cfg)
  

• Prepare data

  Discover GMTSAR output (stack_mode merge/<julian_pair>/, or p2p gmtsar/<ref>_<sec>/merge/) and write the MintPy config. For p2p output it stages the merged unwrap_ll.grd/corr_ll.grd into the <pair>/unwrap_ll.grd shape MintPy expects (symlinked, no multi-GB copies), keeping GMTSAR's Julian yyyyddd_yyyyddd directory naming that prep_gmtsar.py derives pair dates from.

  analyzer.prep_data()
  

• Run

  Run MintPy SBAS time-series analysis. All output is written to workdir/gmtsar_mintpy/.

  analyzer.run()
  

  Parameters:

  Name	Type	Description	Default
  steps	list[str] | None	List of MintPy processing steps to execute. If None, the default full workflow is executed: [ 'load_data', 'modify_network', 'reference_point', 'quick_overview', 'correct_unwrap_error', 'invert_network', 'correct_LOD', 'correct_SET', 'correct_ionosphere', 'correct_troposphere', 'deramp', 'correct_topography', 'residual_RMS', 'reference_date', 'velocity', 'geocode', 'google_earth', 'hdfeos5' ]	None

  Raises:

  Type	Description
  RuntimeError	If tropospheric delay method requires CDS authorization and authorization fails.
  Exception	Propagates exceptions raised during MintPy execution.

• HPC submission

  Inherited from Mintpy_SBAS_Base_Analyzer. Submit the full MintPy run as a single sbatch job (written to workdir/gmtsar_mintpy/mintpy_sbas.sbatch).

  analyzer.submit_hpc()
  

• Clean up

  analyzer.cleanup()
  

  Raises:

  Type	Description
  Exception	Propagates any unexpected errors raised during removal.

Runs GMTSAR's own native SBAS inversion (prep_sbas + the sbas binary) on a GMTSAR_S1 stack_mode stack — no MintPy. Consumes workdir/gmtsar/ and writes cumulative displacement per date (disp_*.grd) and linear velocity (vel.grd) in radar coordinates to workdir/gmtsar_sbas/.

gmtsar_root and gmtsar_env_bin are required here: the sbas binary and gmt come from GMTSAR's own install, not InSARHub's.

Source code in src/insarhub/analyzer/gmtsar_s1_sbas.py

class GMTSAR_SBAS(BaseAnalyzer):
    name = "GMTSAR_SBAS"
    aliases = ("GMTSAR_TS",)   # legacy name
    description = (
        "GMTSAR-native SBAS time-series (prep_sbas + sbas C binary). Consumes "
        "a GMTSAR_S1 stack_mode stack (workdir/gmtsar/), produces disp_*.grd + "
        "vel.grd in workdir/gmtsar_sbas/. No MintPy."
    )
    compatible_processor = "GMTSAR_S1"
    default_config = GMTSAR_SBAS_Config

    def __init__(self, config: GMTSAR_SBAS_Config | None = None):
        super().__init__(config)
        self.config: GMTSAR_SBAS_Config = self.config or GMTSAR_SBAS_Config()
        # Resolved by _resolve_stack_inputs() for a stack-mode run: the raw/
        # whose baseline_table.dat + metadata PRM describe the actual data
        # (not the default first F<N>/raw), and the radar-coordinate pair dir.
        self._raw_dir: Path | None = None
        self._intf_path: Path | None = None

    # ── paths ──────────────────────────────────────────────────────────────
    @property
    def workdir(self) -> Path:
        # absolute: subprocesses run with cwd=sbas_dir, so a relative workdir
        # would not resolve from there
        return Path(self.config.workdir).expanduser().resolve()

    @property
    def _gmtsar_paths(self) -> GMTSARPaths:
        return GMTSARPaths(self.workdir)

    @property
    def stack_dir(self) -> Path:
        """Where GMTSAR_S1 stack_mode wrote its output."""
        return self._gmtsar_paths.case_dir

    @property
    def sbas_dir(self) -> Path:
        return self._gmtsar_paths.sbas_dir

    @staticmethod
    def _unset(v) -> bool:
        """A config value the user didn't set -- treat as 'auto-detect'."""
        return v is None or str(v).strip().lower() in ("", "auto", "none")

    def _subprocess_env(self) -> dict:
        """PATH-prepend GMTSAR bin + its conda env bin, same mechanism as
        GMTSAR_S1._subprocess_env() (see that docstring for the real bug).

        gmtsar_root / gmtsar_env_bin are auto-detected (not user input): an
        explicit config value wins, otherwise _find_gmtsar_root / _find_gmtsar_
        env_bin locate them from $GMTSAR, a GMTSAR script on PATH, or the
        container's own install."""
        from insarhub.processor.gmtsar_s1 import (
            _find_gmtsar_root, _find_gmtsar_env_bin,
        )
        cfg = self.config
        root = None if self._unset(cfg.gmtsar_root) else Path(cfg.gmtsar_root)
        env_bin = None if self._unset(cfg.gmtsar_env_bin) else Path(cfg.gmtsar_env_bin)
        root = _find_gmtsar_root(root)
        env_bin = _find_gmtsar_env_bin(env_bin)

        env = os.environ.copy()
        env["GMTSAR"] = str(root)
        env["PATH"] = ":".join([str(env_bin), str(Path(root) / "bin"),
                                env.get("PATH", "")])
        return env

    # ── network pruning ────────────────────────────────────────────────────
    def _nan_fraction(self, grd: Path) -> float:
        """Fraction of NaN pixels in a .grd (netCDF)."""
        try:
            from netCDF4 import Dataset
            import numpy as np
            with Dataset(grd) as ds:
                for name in ("z", "Band1"):
                    if name in ds.variables:
                        a = ds.variables[name][:]
                        return float(np.ma.getmaskarray(a).mean()
                                     if np.ma.isMaskedArray(a) else np.isnan(a).mean())
        except Exception as exc:
            logger.warning("could not read %s (%s); assuming usable", grd, exc)
        return 0.0

    @staticmethod
    def _baseline_stem_to_julian(baseline: Path) -> dict[str, int]:
        """Map baseline_table.dat col1 (measurement stem) -> int(col2) Julian id.

        prep_sbas keys its ``grep <stem> baseline_table.dat`` on col1, so the
        intf.in it reads must name MEASUREMENT stems (``s1a-iw2-slc-vv-...-005``),
        not the processor's aligned stems (``S1_<date>_ALL_F<N>``) which grep
        never matches -- a mismatch that left ref_id/rep_id empty and prep_sbas
        producing garbage rows. col2 carries the Julian id GMTSAR names the
        pair dirs with.
        """
        out: dict[str, int] = {}
        for ln in baseline.read_text().splitlines():
            parts = ln.split()
            if len(parts) >= 2:
                try:
                    out[parts[0]] = int(float(parts[1]))
                except ValueError:
                    continue
        return out

    def _write_stack_intf_in(self, intf_dir: Path, baseline: Path) -> Path | None:
        """intf.in in prep_sbas's ``ref:rep`` form (measurement stems), built
        from the pair dirs present under intf_dir and the baseline table."""
        jul2stem = {j: s for s, j in self._baseline_stem_to_julian(baseline).items()}
        lines: list[str] = []
        for d in sorted(p for p in intf_dir.iterdir() if p.is_dir()):
            try:
                j1, j2 = (int(x) for x in d.name.split("_"))
            except ValueError:
                continue
            if j1 in jul2stem and j2 in jul2stem:
                lines.append(f"{jul2stem[j1]}:{jul2stem[j2]}")
            else:
                logger.warning("GMTSAR_SBAS: pair %s not in baseline_table.dat; "
                               "skipped", d.name)
        if not lines:
            return None
        out = self.sbas_dir / "intf.in"
        out.write_text("\n".join(lines) + "\n")
        return out

    def _prune_network(self, intf_in: Path, intf_path: Path,
                       baseline: Path) -> Path | None:
        """Drop decorrelated pairs, then any date left orphaned, then keep
        only the largest connected component -- writing a pruned intf.in.

        GMTSAR's sbas solves for every date using pixels valid in EVERY
        interferogram, and does no filtering of its own: a single badly
        decorrelated pair (e.g. a snow-melt/vegetation season crossing) nulls
        out nearly the whole velocity map. Confirmed on real data: 4 of 27
        pairs were ~91% NaN and left only 0.5% of pixels valid; dropping them
        (and the 2 dates they alone connected) took that to 63%.
        """
        import networkx as nx
        cfg = self.config
        stem2jul = self._baseline_stem_to_julian(baseline)
        lines = [l.strip() for l in intf_in.read_text().splitlines() if l.strip()]

        kept, dropped = [], []
        for line in lines:
            ref, rep = line.split(":")
            rid, pid = stem2jul.get(ref), stem2jul.get(rep)
            if rid is None or pid is None:
                dropped.append((line, 1.0))
                continue
            grd = intf_path / f"{rid}_{pid}" / cfg.phase_grd
            frac = self._nan_fraction(grd) if grd.exists() else 1.0
            (dropped if frac > cfg.max_nan_fraction else kept).append((line, frac))

        if not dropped:
            print(f"  network: all {len(lines)} pairs pass "
                  f"(NaN <= {cfg.max_nan_fraction:.0%})")
            return None
        for line, frac in dropped:
            print(f"  dropped {line}  ({frac:.0%} NaN > {cfg.max_nan_fraction:.0%})")

        # keep the largest connected component so sbas isn't singular
        g = nx.Graph()
        g.add_edges_from(tuple(l.split(":")) for l, _ in kept)
        if g.number_of_nodes():
            main = max(nx.connected_components(g), key=len)
            before = len(kept)
            kept = [(l, f) for l, f in kept
                    if set(l.split(":")) <= main]
            if len(kept) < before:
                print(f"  dropped {before - len(kept)} more pair(s) outside the "
                      f"largest connected component")

        if not kept:
            raise RuntimeError(
                "auto_prune removed every pair -- raise max_nan_fraction "
                f"(currently {cfg.max_nan_fraction}) or fix the interferograms.")

        scenes = sorted({s for l, _ in kept for s in l.split(":")})
        # resolve(): prep_sbas runs with cwd=sbas_dir, so a relative path here
        # (from a relative workdir) would not resolve from there
        out = self.sbas_dir.resolve() / "intf_pruned.in"
        out.write_text("\n".join(l for l, _ in kept) + "\n")
        print(f"  network: kept {len(kept)}/{len(lines)} pairs, "
              f"{len(scenes)} dates -> {out.name}")
        self._kept_scenes = {stem2jul[s] for s in scenes}
        return out

    # ── pipeline ───────────────────────────────────────────────────────────
    def _resolve_stack_inputs(self) -> tuple[Path, Path, Path, str, str] | None:
        """(intf.in, baseline_table.dat, radar_intf_dir, phase_grd, corr_grd)
        for a stack_mode run, or None when this isn't a stack-mode layout.

        Detects the radar-coordinate pair dir (per-pair unwrap.grd/corr.grd)
        and the matching raw/ (baseline_table.dat + metadata PRM) across the
        three layouts GMTSAR_S1 stack_mode leaves behind:

            flat                 gmtsar/intf_all   + gmtsar/raw
            subswath, AOI-in-one F<N>/intf_all     + F<N>/raw
            subswath, merged     gmtsar/merge      + F<N>/raw   (true multi)

        For the single-subswath case merge/ only holds geocoded *_ll.grd
        symlinks (no radar unwrap.grd), so the radar check skips it and falls
        through to the F<N>/intf_all that actually produced the stack -- the
        same subswath narrowing GMTSAR_Mintpy_SBAS's _meta_raw_dir does. intf.in
        is (re)written with measurement stems because prep_sbas greps
        baseline_table.dat by col1 (see _write_stack_intf_in).
        """
        p = self._gmtsar_paths
        phase, corr = self.config.phase_grd, self.config.corr_grd

        candidates: list[tuple[Path, Path]] = []
        # merged radar product first (true multi), then each subswath, then flat.
        candidates.append((p.merge_dir, p.meta_raw_dir))
        for d in p._swath_dirs():
            candidates.append((d / "intf_all", d / "raw"))
        candidates.append((p.intf_all_dir, p.raw_dir))

        for intf_dir, raw_dir in candidates:
            baseline = raw_dir / "baseline_table.dat"
            if not baseline.exists():
                continue
            pairs = ([x for x in intf_dir.iterdir() if x.is_dir()]
                     if intf_dir.is_dir() else [])
            if not any((d / phase).exists() for d in pairs):
                continue
            intf_in = self._write_stack_intf_in(intf_dir, baseline)
            if intf_in is None:
                continue
            self._raw_dir = raw_dir
            self._intf_path = intf_dir
            return (intf_in.resolve(), baseline.resolve(), intf_dir.resolve(),
                    phase, corr)
        return None

    def _resolve_sbas_inputs(self) -> tuple[Path, Path, Path, str, str]:  # noqa: D401
        """Return (intf.in, baseline_table.dat, product_dir, phase_name,
        corr_name) for prep_sbas, from EITHER a stack_mode stack or p2p output.

        stack_mode gives native radar-coord grids on a common grid (its whole
        point); p2p geocodes each pair separately, so for p2p we fall back to
        the geocoded *_ll grids staged onto a common grid -- the exact same
        inputs GMTSAR_Mintpy_SBAS feeds MintPy, run through GMTSAR's own sbas
        instead. All resolve()d: prep_sbas runs with cwd=sbas_dir."""
        cfg = self.config
        stack_inputs = self._resolve_stack_inputs()
        if stack_inputs is not None:
            intf_in, baseline, intf_path, phase, corr = stack_inputs
            if cfg.auto_prune:
                intf_in = self._prune_network(intf_in, intf_path, baseline) or intf_in
            return intf_in, baseline, intf_path, phase, corr
        return self._resolve_sbas_inputs_p2p()

    def _resolve_sbas_inputs_p2p(self) -> tuple[Path, Path, Path, str, str]:
        """p2p → sbas inputs. Reuses GMTSAR_Mintpy_SBAS's proven p2p staging
        (per-pair geocoded grids clipped to a common grid) and baseline builder,
        then writes intf.in in prep_sbas's `ref_stem:rep_stem` form from the
        baseline table (prep_sbas keys the pair dir off int(col2)=Julian, which
        is exactly how the staged dirs are named)."""
        from insarhub.analyzer.gmtsar_mintpy_s1_sbas import GMTSAR_Mintpy_SBAS
        from insarhub.config.defaultconfig import GMTSAR_Mintpy_SBAS_Config

        helper = GMTSAR_Mintpy_SBAS(GMTSAR_Mintpy_SBAS_Config(workdir=str(self.workdir)))
        intf = helper._p2p_product_dir()
        if intf is None:
            raise FileNotFoundError(
                f"No stack_mode intf.in under {self.stack_dir} and no p2p "
                f"merge/*_ll.grd either -- run GMTSAR_S1 first (stack_mode or p2p)."
            )
        clip = helper._consistent_intf_dir(intf)

        baseline = helper._gmtsar_paths.baseline_table_auto
        if not baseline.exists():
            if not helper._build_baseline_table(baseline):
                raise FileNotFoundError(
                    "could not build baseline_table.dat from the p2p per-date PRMs "
                    "(need one .PRM + .LED per date under gmtsar/<case>/F<n>/raw/)."
                )

        # Julian id (int of col2) -> stem (col1), to write intf.in `ref:rep`.
        jul2stem: dict[int, str] = {}
        for ln in baseline.read_text().splitlines():
            p = ln.split()
            if len(p) >= 2:
                try:
                    jul2stem[int(float(p[1]))] = p[0]
                except ValueError:
                    continue
        lines = []
        for d in sorted(p for p in clip.iterdir() if p.is_dir()):
            try:
                j1, j2 = (int(x) for x in d.name.split("_"))
            except ValueError:
                continue
            if j1 in jul2stem and j2 in jul2stem:
                lines.append(f"{jul2stem[j1]}:{jul2stem[j2]}")
            else:
                logger.warning("p2p sbas: pair %s not in baseline_table; skipped", d.name)
        if not lines:
            raise RuntimeError(
                "no p2p pairs matched the baseline table -- Julian dir names and "
                "baseline_table.dat dates disagree.")
        intf_in = self.sbas_dir / "intf.in"
        intf_in.write_text("\n".join(lines) + "\n")
        print(f"{Fore.CYAN}p2p: {len(lines)} pair(s) -> GMTSAR-native sbas on the "
              f"geocoded common grid ({clip}){Fore.RESET}")
        # p2p pairs are geocoded; auto_prune's radar-stem logic doesn't apply.
        # Also tells run() to skip the radar->ll projection: sbas ran on lon/lat
        # grids, so vel.grd/disp_*.grd come out geographic already.
        self._p2p_geocoded = True
        return (intf_in.resolve(), baseline.resolve(), clip.resolve(),
                "unwrap_ll.grd", "corr_ll.grd")

    def prep_data(self) -> str:
        """Run prep_sbas → intf.tab + scene.tab in sbas_dir. Returns the
        sbas command line prep_sbas echoes (with N/S/xdim/ydim filled in)."""
        self.sbas_dir.mkdir(parents=True, exist_ok=True)
        intf_in, baseline, intf_path, phase_name, corr_name = self._resolve_sbas_inputs()
        # prep_sbas intf.in baseline_table.dat <intf_path> <phase_grd> <corr_grd>
        cmd = ["prep_sbas", str(intf_in), str(baseline), str(intf_path),
               phase_name, corr_name]
        proc = subprocess.run(cmd, cwd=str(self.sbas_dir), capture_output=True,
                              text=True, env=self._subprocess_env())
        if proc.returncode != 0:
            raise RuntimeError(f"prep_sbas failed:\n{proc.stdout}\n{proc.stderr}")
        # prep_sbas prints:  sbas intf.tab scene.tab <N> <S> <xdim> <ydim>
        m = re.search(r"^sbas\s+intf\.tab\s+scene\.tab\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)",
                      proc.stdout, re.MULTILINE)
        if not m:
            raise RuntimeError(
                f"prep_sbas did not echo an sbas command line:\n{proc.stdout}")
        n, s, xdim, ydim = (int(v) for v in m.groups())

        # prep_sbas always rebuilds scene.tab from the FULL baseline_table, so
        # after pruning it still lists dropped dates -- leaving them in makes
        # sbas's least-squares system singular for those (orphan) columns.
        kept = getattr(self, "_kept_scenes", None)
        if kept:
            tab = self.sbas_dir / "scene.tab"
            rows = [l for l in tab.read_text().splitlines() if l.strip()]
            keep_rows = [l for l in rows if int(l.split()[0]) in kept]
            if len(keep_rows) != len(rows):
                tab.write_text("\n".join(keep_rows) + "\n")
                print(f"  scene.tab: {len(rows)} -> {len(keep_rows)} dates "
                      f"(dropped orphaned dates)")
            s = len(keep_rows)
        n = len([l for l in (self.sbas_dir / "intf.tab").read_text().splitlines() if l.strip()])
        return f"sbas intf.tab scene.tab {n} {s} {xdim} {ydim}"

    def _resolve_geometry(self) -> tuple[float, float]:
        """sbas's -range and -incidence for this scene: (metres, degrees).

        Both feed one term in sbas -- ``scale = 4*pi / wl / rng / sin(theta)``
        (sbas.c), which forms the DEM-error column of the design matrix
        (``G[...] = bperp[i] * scale`` in sbas_utils.c). Neither scales the
        displacement time series itself, which is why GMTSAR's recipe calls
        the incidence value "largely irrelevant" -- but both are scene
        geometry that the super-master PRM already knows exactly, so neither
        is guessed.

        -range follows the recipe's formula (§12b):

            Range = ({[(c / rng_samp_rate) / 2] * ((x_min + x_max) / 2)} / 2)
                    + near_range

        -incidence is the standard spherical-earth conversion from that
        range, with Rs = earth_radius + SC_height:

            cos(look) = (Rs^2 + R^2 - Re^2) / (2 * Rs * R)
            sin(inc)  = Rs * sin(look) / Re

        Measured on a real 3-subswath p100_f466 frame this gives 39.53 deg at
        mid-swath, against sbas's built-in default of 37 (~6% off on that
        term) and the recipe's own hardcoded 40 (~1% off) -- so the recipe's
        choice is about right for Sentinel-1 IW while sbas's default is not,
        and computing it is exact for any frame or sensor. Validated against
        the published Sentinel-1 IW geometry: evaluating the same conversion
        at the merged grid's near and far edges gives 30.49 and 45.98 deg,
        reproducing the documented ~29.1-46.0 deg IW1-IW3 span.

        Falls back to config.range_dist / config.incidence with a warning if
        the PRM can't be read.
        """
        cfg = self.config
        try:
            raw = self._raw_dir or self._gmtsar_paths.meta_raw_dir
            prm = next(iter(sorted(raw.glob("S1_*_ALL_F*.PRM"))), None)
            if prm is None:
                raise FileNotFoundError("no super-master PRM")
            txt = prm.read_text()

            def _prm(key: str) -> float:
                m = re.search(rf"(?m)^{key}\s*=\s*(\S+)", txt)
                if not m:
                    raise KeyError(key)
                return float(m.group(1))

            rng_samp_rate = _prm("rng_samp_rate")
            near_range = _prm("near_range")

            intf_path = self._intf_path or self._gmtsar_paths.product_dir()
            grd = next(iter(sorted(intf_path.glob(f"*/{cfg.phase_grd}"))), None)
            if grd is None:
                raise FileNotFoundError(f"no {cfg.phase_grd} to size the grid from")
            info = subprocess.run(["gmt", "grdinfo", "-C", str(grd)],
                                  capture_output=True, text=True,
                                  env=self._subprocess_env())
            fields = info.stdout.split()
            x_min, x_max = float(fields[1]), float(fields[2])

            # NB: the recipe's prose and its worked example disagree. Written
            # out it reads "({[(c/rng_samp_rate)/2] * ((x_min+x_max)/2)} / 2)
            # + near_range", but its own numbers are
            # "({[3e8/64345238.125714/2] * 47704} / 2) + 845481.851848 =
            # 901,085" -- i.e. it multiplies by x_max and halves that, not by
            # the midpoint and halves again. Taking the prose literally gives
            # 873,283 for the same scene, 28 km low. The worked example is
            # the physically sensible one and is what's implemented here:
            #   range = (slant range per sample) * (midpoint sample) + near_range
            # Exact c is used where the recipe writes "~3 x 10^8 m/s"
            # (a 0.004% difference).
            c = 299792458.0
            rng = ((c / rng_samp_rate) / 2.0) * ((x_min + x_max) / 2.0) + near_range

            earth_radius = _prm("earth_radius")
            sc_height = _prm("SC_height")
            rs = earth_radius + sc_height
            look = math.acos((rs ** 2 + rng ** 2 - earth_radius ** 2) / (2 * rs * rng))
            inc = math.degrees(math.asin(rs * math.sin(look) / earth_radius))

            print(f"  -range {rng:.0f} m  -incidence {inc:.2f} deg  "
                  f"(computed from {prm.name}: rng_samp_rate={rng_samp_rate:.6g}, "
                  f"near_range={near_range:.6g}, Re={earth_radius:.0f}, "
                  f"H={sc_height:.0f}, x={x_min:.0f}..{x_max:.0f})")
            return rng, inc
        except Exception as e:
            logger.warning(
                "Could not compute sbas -range/-incidence from the PRM (%s); falling "
                "back to config.range_dist=%s, config.incidence=%s. Both are "
                "frame-specific -- verify them, or set them explicitly.",
                e, cfg.range_dist, cfg.incidence)
            return float(cfg.range_dist), float(cfg.incidence)

    def _run_via_container(self) -> None:
        """Re-invoke `insarhub analyzer ... run` inside self.config.container,
        which ships GMTSAR + insarhub. Mirrors Mintpy_SBAS_Base._run_via_container."""
        import subprocess
        from insarhub.utils.container import wrap_container_cmd
        cli_cmd = f"insarhub analyzer -N {type(self).name} -w {self.workdir} run --step sbas"
        wrapped = wrap_container_cmd(self.config.container, cli_cmd, Path(self.workdir))
        result = subprocess.run(wrapped, shell=True)
        if result.returncode != 0:
            raise RuntimeError(f"Container run failed (exit {result.returncode}): {wrapped}")

    def run(self, steps=None) -> None:
        """prep_data() (if needed) then run the sbas inversion with the
        configured flags → disp_*.grd + vel.grd in sbas_dir."""
        if self.config.container and not os.environ.get("INSARHUB_CONTAINER_CHILD"):
            return self._run_via_container()
        sbas_base = self.prep_data()   # "sbas intf.tab scene.tab N S xdim ydim"
        # Stamp the folder AFTER prep_data(), not before: prep_data() builds a
        # GMTSAR_Mintpy_SBAS helper on this same workdir to reuse its p2p
        # staging, and that helper's __init__ stamps the folder as
        # "GMTSAR_Mintpy_SBAS". Marking here overwrites it back to the analyzer
        # the user actually ran, so the GUI badge names the right one.
        from insarhub.utils.tool import write_workflow_marker
        write_workflow_marker(self.workdir, analyzer=type(self).name)
        print(f"prep_sbas OK -> {sbas_base}")
        cfg = self.config
        cmd = sbas_base.split()
        if cfg.smooth:
            cmd += ["-smooth", str(cfg.smooth)]
        if cfg.atm_iters:
            cmd += ["-atm", str(cfg.atm_iters)]
        rng, inc = self._resolve_geometry()
        cmd += ["-wavelength", str(cfg.wavelength),
                "-incidence", f"{inc:.2f}",
                "-range", f"{rng:.0f}"]
        if cfg.rms:
            cmd += ["-rms"]
        if cfg.dem_err:
            cmd += ["-dem"]

        log = self.sbas_dir / "sbas.log"
        print(f"Running SBAS inversion: {' '.join(cmd)}\n  (log: {log})")
        # stream sbas's own progress lines live (tee to console + log) instead
        # of burying them in the log until the end
        with open(log, "w") as lf:
            proc = subprocess.Popen(cmd, cwd=str(self.sbas_dir),
                                    stdout=subprocess.PIPE,
                                    stderr=subprocess.STDOUT,
                                    text=True, bufsize=1,
                                    env=self._subprocess_env())
            for line in proc.stdout:
                print(f"  [sbas] {line.rstrip()}", flush=True)
                lf.write(line)
            proc.wait()
        if proc.returncode != 0:
            raise RuntimeError(f"sbas failed (rc={proc.returncode}) -- see {log}")
        n_disp = len(list(self.sbas_dir.glob("disp_*.grd")))
        print(f"SBAS inversion complete: {self.sbas_dir}\n"
              f"  vel.grd + {n_disp} disp_*.grd (cumulative displacement per date)")
        logger.info("SBAS inversion complete: %s (vel.grd, disp_*.grd)", self.sbas_dir)
        if getattr(self, "_p2p_geocoded", False):
            self._alias_geocoded_outputs()
        else:
            self.geocode()

    def _alias_geocoded_outputs(self) -> None:
        """p2p sbas ran on geocoded (_ll) grids, so vel.grd/disp_*.grd are
        ALREADY in lon/lat -- no proj_ra2ll needed. Write *_ll.grd aliases so
        anything expecting GMTSAR's geocoded naming (the results viewer, the
        recipe's Google-Earth step) finds them."""
        print(f"{Fore.CYAN}p2p: sbas ran on geocoded grids -- vel.grd/disp_*.grd "
              f"are already lon/lat; writing *_ll.grd aliases.{Fore.RESET}")
        for grd in ([self.sbas_dir / "vel.grd"]
                    + sorted(self.sbas_dir.glob("disp_*.grd"))):
            if grd.exists():
                out = grd.with_name(f"{grd.stem}_ll.grd")
                if not out.exists():
                    out.symlink_to(grd.name)

    def geocode(self) -> None:
        """Project sbas's radar-coordinate output into lat/lon (recipe §12c).

        sbas writes vel.grd and disp_*.grd in RADAR coordinates, which can't
        be mapped or compared against anything geographic until projected --
        the recipe's own last step, and the reason its results are viewable
        in Google Earth. Needs trans.dat, which the merge stage produced
        (mergeprep, for a multi-subswath stack); linked in rather than copied,
        as the recipe does, since it can be well over a gigabyte.

        Non-fatal: the inversion itself has already succeeded and its radar
        grids are on disk by this point, so a missing trans.dat degrades this
        to a warning rather than throwing away the run.
        """
        env = self._subprocess_env()
        candidates = [self._gmtsar_paths.merge_dir / "trans.dat"]
        if self._raw_dir is not None:
            candidates.append(self._raw_dir.parent / "topo" / "trans.dat")
        candidates.append(self._gmtsar_paths.topo_dir / "trans.dat")
        trans_src = next((c for c in candidates if c.exists()), None)
        if trans_src is None:
            logger.warning(
                "No trans.dat found (looked in %s) -- skipping geocoding; "
                "vel.grd/disp_*.grd remain in radar coordinates.",
                ", ".join(str(c) for c in candidates))
            return

        trans = self.sbas_dir / "trans.dat"
        if not trans.exists():
            trans.symlink_to(os.path.relpath(trans_src, self.sbas_dir))

        targets = [p for p in ([self.sbas_dir / "vel.grd"]
                               + sorted(self.sbas_dir.glob("disp_*.grd"))) if p.exists()]
        done = 0
        for grd in targets:
            out = grd.with_name(f"{grd.stem}_ll.grd")
            if out.exists():
                done += 1
                continue
            # proj_ra2ll.csh caches raln.grd/ralt.grd and will not regenerate
            # them, which is what makes projecting many grids cheap after the
            # first -- but also why a stale pair has to be cleared by hand.
            r = subprocess.run(["proj_ra2ll.csh", "trans.dat", grd.name, out.name],
                               cwd=str(self.sbas_dir), capture_output=True,
                               text=True, env=env)
            if r.returncode == 0 and out.exists():
                done += 1
            else:
                logger.warning("proj_ra2ll failed for %s: %s",
                               grd.name, (r.stderr or r.stdout).strip()[:300])

        vel_ll = self.sbas_dir / "vel_ll.grd"
        if vel_ll.exists():
            cpt = self.sbas_dir / "vel_ll.cpt"
            with open(cpt, "w") as fh:
                subprocess.run(["gmt", "grd2cpt", vel_ll.name, "-Z", "-Cjet"],
                               cwd=str(self.sbas_dir), stdout=fh,
                               stderr=subprocess.DEVNULL, env=env)
            if cpt.stat().st_size:
                subprocess.run(["grd2kml.csh", "vel_ll", cpt.name],
                               cwd=str(self.sbas_dir), capture_output=True, env=env)
        print(f"  geocoded {done}/{len(targets)} grid(s) -> *_ll.grd"
              + (" (+ vel_ll.kml)" if (self.sbas_dir / "vel_ll.kml").exists() else ""))

    def extract_time_series(self, lon: float, lat: float,
                            m_rng: int = 5, m_azi: int = 5) -> Path:
        """Point time series via GMTSAR's extract_one_time_series →
        time_series.dat in sbas_dir. Needs a PRM + dem.grd + scene.tab."""
        scene_tab = self.sbas_dir / "scene.tab"
        dem = self._gmtsar_paths.dem_grd
        raw = self._raw_dir or self._gmtsar_paths.raw_dir
        prm = next(raw.glob("S1_*.PRM"), None)
        if prm is None:
            raise FileNotFoundError("no S1_*.PRM in stack raw/ for llt2rat")
        cmd = ["extract_one_time_series", str(lon), str(lat), str(prm),
               str(dem), str(scene_tab), str(m_rng), str(m_azi)]
        subprocess.run(cmd, cwd=str(self.sbas_dir), check=True,
                       env=self._subprocess_env())
        return self.sbas_dir / "time_series.dat"

Usage

• Create Analyzer

  from insarhub import Analyzer
  
  analyzer = Analyzer.create('GMTSAR_SBAS', workdir='/your/work/dir',
                             gmtsar_root='/path/to/gmtsar',
                             gmtsar_env_bin='/path/to/conda/envs/gmtsar/bin')
  

  OR with explicit config:

  from insarhub.config.defaultconfig import GMTSAR_SBAS_Config
  
  cfg = GMTSAR_SBAS_Config(
      workdir='/your/work/dir',
      gmtsar_root='/path/to/gmtsar',
      gmtsar_env_bin='/path/to/conda/envs/gmtsar/bin',
  )
  analyzer = Analyzer.create('GMTSAR_SBAS', config=cfg)
  

• Prepare data

  Build intf.tab and scene.tab from the stack's baseline_table.dat, then echo the sbas intf.tab scene.tab N S xdim ydim command line to run.

  analyzer.prep_data()
  

• Run

  Run the sbas inversion, streaming its progress to the console and sbas.log under workdir/gmtsar_sbas/.

  analyzer.run()
  

Runs dolphin's timeseries.run on the unwrapped stack from the ISCE3_Burst processor (Sentinel-1 bursts), writing to workdir/timeseries/. The NISAR counterpart is ISCE3_Dolphin_NISAR_PL; both inherit the inversion from Dolphin_PL_Base_Analyzer.

Water is excluded by default (apply_water_mask=True) using the processor's dem/water_mask.tif. Turn it off to invert open water too.

Legacy names

ISCE3_Dolphin_PL, ISCE3_Dolphin_TS, Dolphin_TS and Dolphin_SBAS all still resolve to this analyzer, so saved insarhub_config.json files and older CLI commands keep working. They are hidden from the analyzer list. The same applies to the config class: ISCE3_Dolphin_PL_Config and ISCE3_Dolphin_PL_S1_Config are aliases of ISCE3_Dolphin_S1_PL_Config.

Source code in src/insarhub/analyzer/isce3_dolphin_s1_pl.py

class ISCE3_Dolphin_S1_PL(Dolphin_PL_Base_Analyzer):
    """Dolphin time-series over an ISCE3_Burst (Sentinel-1) stack.

    Everything is inherited from the base: the C-band wavelength is a plain
    config default (``ISCE3_Dolphin_S1_PL_Config.wavelength``), so the base's
    :meth:`~insarhub.analyzer.dolphin_base.Dolphin_PL_Base_Analyzer._wavelength`
    finds it without an override. See
    :class:`~insarhub.analyzer.isce3_dolphin_NISAR_PL.ISCE3_Dolphin_NISAR_PL`
    for the case that does need one.
    """

    name = "ISCE3_Dolphin_S1_PL"
    #: Legacy names, kept resolvable so saved insarhub_config.json files and CLI
    #: commands from before the per-sensor split keep working. ``ISCE3_Dolphin_PL``
    #: belongs here rather than on the NISAR child: until the split it WAS the
    #: Sentinel-1 analyzer (C-band wavelength, water mask, LOS projection), so an
    #: old config invoked by that name must keep landing on this class.
    aliases = ("Dolphin_SBAS", "Dolphin_TS", "ISCE3_Dolphin_TS",
               "ISCE3_Dolphin_PL")
    description = ("Time-series inversion of an ISCE3_Burst (Sentinel-1) stack "
                   "with dolphin: cumulative displacement per date, velocity, residuals.")
    # One analyzer per upstream, each carrying its own config -- the same shape
    # as the Mintpy family (Hyp3_Mintpy_SBAS / ISCE2_Mintpy_SBAS / ...). This
    # used to be a single analyzer with a ("ISCE3_Burst", "ISCE3_NISAR") tuple
    # and one shared config, but default_config is a per-class binding and
    # nothing dispatches on the actual upstream, so a NISAR stack silently got
    # the Sentinel-1 C-band wavelength.
    compatible_processor = "ISCE3_Burst"
    default_config = ISCE3_Dolphin_S1_PL_Config

Usage

• Create Analyzer

  from insarhub import Analyzer
  
  analyzer = Analyzer.create('ISCE3_Dolphin_S1_PL', workdir='/your/work/dir')
  

  OR with explicit config:

  from insarhub.config.defaultconfig import ISCE3_Dolphin_S1_PL_Config
  
  cfg = ISCE3_Dolphin_S1_PL_Config(workdir='/your/work/dir')
  analyzer = Analyzer.create('ISCE3_Dolphin_S1_PL', config=cfg)
  

• Run

  Run the dolphin time-series inversion (cumulative displacement, velocity, residuals) for the stack in this workdir.

  analyzer.run()
  

The NISAR counterpart of ISCE3_Dolphin_S1_PL — same timeseries.run, same products, same workdir/timeseries/ output — consuming the stack from the ISCE3_NISAR processor.

Three differences, each forced by what ISCE3_NISAR produces:

• Wavelength is read from the GSLC metadata rather than pinned; NISAR is L-band and its frequency A/B bands differ. Set wavelength to override.
• apply_water_mask defaults to False — ISCE3_NISAR runs no dem stage, so there is no mask to apply.
• los_projection is hidden — 'vertical' needs the processor's static stage, which ISCE3_NISAR does not run.

nisar_frequency / nisar_polarization must match what the processor phase-linked.

Legacy name

ISCE3_Dolphin_PL_NISAR still resolves to this analyzer, and ISCE3_Dolphin_PL_NISAR_Config is an alias of ISCE3_Dolphin_NISAR_PL_Config.

Source code in src/insarhub/analyzer/isce3_dolphin_nisar_pl.py

class ISCE3_Dolphin_NISAR_PL(Dolphin_PL_Base_Analyzer):

    """Dolphin time-series over a NISAR GSLC stack (ISCE3_NISAR).

    Identical inversion to the Sentinel-1 analyzer -- same dolphin
    ``timeseries.run``, same products -- so everything comes from the shared
    Dolphin_PL_Base_Analyzer. What changes is the config bound to it and how
    the wavelength is obtained:
    NISAR is L-band and its two frequency groups differ, so the value is read
    from the GSLC metadata instead of being pinned to a constant.
    """
    name = "ISCE3_Dolphin_NISAR_PL"
    #: Only this class's OWN former name. Registry.register() maps every alias to
    #: the registering class, so listing the Sentinel-1 analyzer's legacy names
    #: ("Dolphin_SBAS", "Dolphin_TS", "ISCE3_Dolphin_TS", "ISCE3_Dolphin_PL")
    #: here would silently re-point them at this NISAR class -- an old S1 config
    #: invoked by a legacy name would then be built with the NISAR config and
    #: fail on a missing wavelength. Those belong to ISCE3_Dolphin_S1_PL.
    #: Inheriting from the sensor-neutral base rather than from the S1 analyzer
    #: is what keeps that separation structural instead of a comment.
    aliases = ("ISCE3_Dolphin_PL_NISAR",)
    description = ("Time-series inversion of an ISCE3_NISAR (NISAR GSLC) stack "
                   "with dolphin: cumulative displacement per date, velocity, residuals.")
    compatible_processor = "ISCE3_NISAR"
    default_config = ISCE3_Dolphin_NISAR_PL_Config

    #: Where a NISAR GSLC keeps its centre frequency, in preference order.
    #: ``{f}`` is the frequency group letter from the config.
    _CENTER_FREQ_PATHS = (
        "/science/LSAR/GSLC/grids/frequency{f}/centerFrequency",
        "/science/LSAR/GSLC/grids/frequency{f}/processingInformation/centerFrequency",
        "/science/LSAR/identification/centerFrequency",
    )

    _C = 299792458.0        # m/s

    def _gslc_files(self) -> list[Path]:
        """The stack's GSLC granules. Mirrors ISCE3_NISAR's own gslc_dir default
        (workdir/slc) so the analyzer looks where the processor wrote."""
        d = getattr(self.config, "gslc_dir", None)
        d = Path(d).expanduser() if d else self.workdir / "slc"
        return sorted(d.glob("*GSLC*.h5"))

    def _wavelength(self) -> float:
        """Config value if set, else c / centreFrequency from the GSLC metadata.

        Taking the Sentinel-1 C-band constant here would scale every
        displacement by roughly 4.3x -- L-band is ~0.24 m against C-band's
        0.055 m -- and nothing downstream would flag it, so this reads the
        actual product rather than swapping in a second hardcoded number.
        Frequency A and B have different centre frequencies, which is the other
        reason not to pin a constant.
        """
        w = getattr(self.config, "wavelength", None)
        if w:
            return float(w)

        import h5py
        import numpy as np

        freq = str(getattr(self.config, "nisar_frequency", "A")).upper()
        files = self._gslc_files()
        if not files:
            raise ValueError(
                f"{self.name}: no *GSLC*.h5 found to read the wavelength from. "
                f"Set `wavelength` (metres) explicitly, or point `gslc_dir` at "
                f"the granules the ISCE3_NISAR processor used.")

        tried = [q.format(f=freq) for q in self._CENTER_FREQ_PATHS]
        with h5py.File(files[0], "r") as h5:
            for path in tried:
                if path not in h5:
                    continue
                hz = float(np.ravel(h5[path][()])[0])
                if hz <= 0:
                    continue
                lam = self._C / hz
                print(f"[{self.name}] wavelength   : {lam:.6f} m "
                      f"(frequency {freq}, {hz / 1e6:.1f} MHz, from {files[0].name})")
                return lam

        raise ValueError(
            f"{self.name}: no centre frequency in {files[0].name}; looked at "
            f"{', '.join(tried)}. Set `wavelength` (metres) explicitly.")

Usage

• Create Analyzer

  from insarhub import Analyzer
  
  analyzer = Analyzer.create('ISCE3_Dolphin_NISAR_PL', workdir='/your/work/dir')
  

  OR with explicit config:

  from insarhub.config.defaultconfig import ISCE3_Dolphin_NISAR_PL_Config
  
  cfg = ISCE3_Dolphin_NISAR_PL_Config(workdir='/your/work/dir', nisar_frequency='A')
  analyzer = Analyzer.create('ISCE3_Dolphin_NISAR_PL', config=cfg)
  

• Run

  Run the dolphin time-series inversion (cumulative displacement, velocity, residuals) for the stack in this workdir.

  analyzer.run()