分析器

InSARHub 分析器模块提供 InSAR 时序分析工作流。

导入分析器

导入 Analyzer 类以访问所有时序分析功能
```
from insarhub import Analyzer
```
查看可用分析器

列出所有已注册的分析器
```
Analyzer.available()
```

可用分析器

Mintpy_SBAS_Base_AnalyzerHyp3_Mintpy_SBASISCE2_Mintpy_SBASGMTSAR_Mintpy_SBASGMTSAR_SBASISCE3_Dolphin_S1_PLISCE3_Dolphin_NISAR_PL

InSARHub 将 Mintpy 封装为其分析后端之一。Mintpy_SBAS_Base_Analyzer 基于可复用的基础配置类实现，提供 Mintpy 完整的 smallbaselineApp 逻辑。为用户提供类似于直接使用 MintPy 的体验，支持对处理参数和步骤进行完整自定义。

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}")

使用方法

使用参数创建分析器

初始化分析器实例

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

或

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

或

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)

基础配置 Mintpy_SBAS_Base_Config 包含 Mintpy smallbaselineApp.cfg 的所有参数。有关每个参数的详细说明，请参阅 Mintpy 官方配置文档。

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" 而非固定数值。在 prep_data 期间，InSARHub 会检查堆叠实际的相干性分布，并将每个参数解析写入 .mintpy.cfg：

参数	解析依据	上限
`network_minCoherence`	在保持网络连通且冗余的前提下最严格的阈值	≤ 0.6
`networkInversion_maskThreshold`	保留可靠像素比例的分位数	≤ 0.6
`reference_minCoherence`	第 98 百分位（最低 0.30），用于稳定的参考点	≤ 0.85

仅当数据低于上限时才会启用自适应；干净、高相干的堆叠直接取上限值。将其中任一参数设为明确的数值即可完全覆盖自适应逻辑。

运行

根据提供的配置运行 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 / SLURM 模式）

生成一个涵盖所有选定步骤的单个 sbatch 脚本并提交至 SLURM。Hyp3_Mintpy_SBAS 和 ISCE2_Mintpy_SBAS 均继承此方法。
```
# 将完整流程作为一个 SLURM 作业提交
analyzer.submit_hpc()

# 仅提交特定步骤
analyzer.submit_hpc(steps=["velocity", "geocode"])
```
脚本写入 <workdir>/mintpy/mintpy_sbas.sbatch，作业状态保存至 mintpy/mintpy_job.json。SLURM 资源来自 <workdir>/sbatch_options.json 的 "17" 步骤键 — 与 ISCE2_S1 自身 HPC 提交（步骤 01–16）使用同一个文件，因为处理器和分析器通常共用同一工作目录。默认值：time=24:00:00、ntasks=1、cpus_per_task=16、mem=128G、partition=all。

submit_hpc() 成功时返回 SLURM 作业 ID 字符串；若 sbatch_options.json 刚被创建（或补充了缺失的 "17" 条目），则返回 None — 调用方应检查 None 并停止，而不是将其当作提交成功处理：
```
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 刚被创建/更新 — 请先检查，再重新提交。")
```
直接编辑 sbatch_options.json 中的 "17" 步骤以更改资源（例如 {"17": {"time": "48:00:00", "mem": "256G", "partition": "gpu"}}），然后再次调用 submit_hpc()。
绘图

基于已计算完成的结果，（重新）生成 mintpy/pic/ 下的图片，不重新计算任何内容。run() 自身的自动绘图只在单次调用中涵盖一个以上步骤时才会触发（与 MintPy 自身的 CLI 语义一致）— CLI 和 GUI 在内部都是逐步执行每个步骤以提供逐步进度反馈，因此该条件在那里实际上永远不会触发；plot() 是显式的独立替代方案，两者都在各自的步骤序列完成后调用一次（或按需调用，例如调整了与绘图相关的配置值后，只想重新生成图片而不重新运行整个流程）。
```
analyzer.plot()
```
无需本地安装 MintPy（或 ISCE2）

将 container 字段设置为 Apptainer/Singularity .sif 镜像的路径，或 Docker 镜像引用（name[:tag]），run()/prep_data()/submit_hpc() 都会在容器内而非宿主机上重新执行同一个 insarhub analyzer ... CLI 调用 — 工作目录会以相同路径绑定挂载，因此输出会像本机运行一样落在原处。容器镜像只需在 MintPy（ISCE2_Mintpy_SBAS 还需要 ISCE2）旁额外安装 insarhub（可参考仓库根目录的 docker/dev/Dockerfile.isce2-mintpy 作为现成示例）。
```
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 是按次调用的设置，而非持久化配置 — 之后每次调用若也要在容器内运行，都需要再次设置。

清理

删除时序处理过程中生成的中间处理文件

analyzer.cleanup()

Raises:

Type	Description
`Exception`	Propagates any unexpected errors raised during removal.

Hyp3_Mintpy_SBAS 是专门为处理 HyP3 InSAR 产品时序数据而预配置的分析器，扩展自 Mintpy_SBAS_Base_Analyzer。

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())

使用方法

使用参数创建分析器

初始化分析器实例

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

或

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

或

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)

准备数据

将从 HyP3 服务器下载的干涉图数据准备至 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()

运行

根据提供的配置运行 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.

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()

提交（HPC / SLURM 模式）

继承自 Mintpy_SBAS_Base_Analyzer，将完整 MintPy 流程作为单个 sbatch 作业提交。
```
analyzer.submit_hpc()
```

清理

删除时序处理过程中生成的中间处理文件

analyzer.cleanup()

Raises:

Type	Description
`Exception`	Propagates any unexpected errors raised during removal.

ISCE2_Mintpy_SBAS 分析器扩展自 Mintpy_SBAS_Base_Analyzer，专为 ISCE2 stackSentinel 输出预配置。prep_data() 自动发现 isce/ 目录中的干涉图和几何数据，并将 MintPy 配置写入 mintpy/.mintpy.cfg。所有 MintPy 输出写入 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)

使用方法

创建分析器

from insarhub import Analyzer

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

或使用显式配置：

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)

准备数据

自动发现 ISCE2 输出并写入 mintpy/.mintpy.cfg。
```
analyzer.prep_data()
```
运行

运行 MintPy SBAS 时序分析。所有输出写入 workdir/mintpy/。
```
analyzer.run()
```
提交（HPC / SLURM 模式）

继承自 Mintpy_SBAS_Base_Analyzer，将完整 MintPy 流程作为单个 sbatch 作业提交。
```
analyzer.submit_hpc()
```
清理

删除 load_data 后不再需要的大型 ISCE2 中间目录和输入数据。删除 isce/coarse_interferograms/、isce/ESD/、isce/coreg_secondarys/、isce/interferograms/、slc/ 和 dem/。
```
analyzer.cleanup()
```

对 GMTSAR_S1 处理器生成的堆叠运行 MintPy SBAS，将 GMTSAR 的地理编码 *_ll.grd 产品和 baseline_table.dat 交给 MintPy 的 prep_gmtsar.py 加载器。它是 ISCE2_Mintpy_SBAS 的 MintPy 对应物。输出写入 workdir/gmtsar_mintpy/，独立目录，不会与同一工作目录中的 Hyp3 或 ISCE MintPy 运行相互覆盖。

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}")

使用方法

创建分析器

from insarhub import Analyzer

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

或使用显式配置：

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)

准备数据

发现 GMTSAR 输出（stack_mode 的 merge/<julian_pair>/，或 p2p 的 gmtsar/<ref>_<sec>/merge/）并写入 MintPy 配置。对于 p2p 输出，它将合并后的 unwrap_ll.grd/corr_ll.grd 暂存为 MintPy 期望的 <pair>/unwrap_ll.grd 结构（符号链接，无需复制数 GB 网格），并保留 GMTSAR 的儒略日 yyyyddd_yyyyddd 目录命名，prep_gmtsar.py 据此推导配对日期。
```
analyzer.prep_data()
```

运行

运行 MintPy SBAS 时序分析。所有输出写入 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 / SLURM 模式）

继承自 Mintpy_SBAS_Base_Analyzer，将完整 MintPy 流程作为单个 sbatch 作业提交（脚本写入 workdir/gmtsar_mintpy/mintpy_sbas.sbatch）。
```
analyzer.submit_hpc()
```

清理

analyzer.cleanup()

Raises:

Type	Description
`Exception`	Propagates any unexpected errors raised during removal.

在 GMTSAR_S1 stack_mode 堆叠上运行 GMTSAR 自带的原生 SBAS 反演（prep_sbas + sbas 二进制程序）——不涉及 MintPy。读取 workdir/gmtsar/，在 workdir/gmtsar_sbas/ 下以雷达坐标生成每个日期的累计位移（disp_*.grd）和线性速度（vel.grd）。

此处 gmtsar_root 与 gmtsar_env_bin 均为必需项：sbas 二进制程序和 gmt 来自 GMTSAR 自身的安装，而非 InSARHub。

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"

使用方法

创建分析器

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')

或使用显式配置：

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)

准备数据

从堆叠的 baseline_table.dat 构建 intf.tab 和 scene.tab，然后回显待运行的 sbas intf.tab scene.tab N S xdim ydim 命令行。
```
analyzer.prep_data()
```
运行

运行 sbas 反演，将进度实时输出到控制台及 workdir/gmtsar_sbas/ 下的 sbas.log。
```
analyzer.run()
```

在 ISCE3_Burst 处理器（Sentinel-1 burst）生成的解缠堆叠上运行 dolphin 的 timeseries.run，输出写入 workdir/timeseries/。NISAR 对应版本为 ISCE3_Dolphin_NISAR_PL，两者的反演逻辑均继承自 Dolphin_PL_Base_Analyzer。

默认将水体排除在反演之外（apply_water_mask=True），使用处理器的 dem/water_mask.tif；关闭后开阔水域也会参与反演。

旧名称

ISCE3_Dolphin_PL、ISCE3_Dolphin_TS、Dolphin_TS 与 Dolphin_SBAS 均仍解析到本分析器，因此已保存的 insarhub_config.json 和旧的 CLI 命令继续可用；它们不会出现在分析器列表中。配置类同理：ISCE3_Dolphin_PL_Config 与 ISCE3_Dolphin_PL_S1_Config 是 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

使用方法

创建分析器

from insarhub import Analyzer

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

或使用显式配置：

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)

运行

对该 workdir 中的堆叠运行 dolphin 时间序列反演（每个日期的累计位移、速度、残差）。
```
analyzer.run()
```

ISCE3_Dolphin_S1_PL 的 NISAR 对应版本——相同的 timeseries.run、相同的产品、相同的 workdir/timeseries/ 输出——消费 ISCE3_NISAR 处理器生成的堆叠。

三处差异，均由 ISCE3_NISAR 的产出决定：

波长从 GSLC 元数据读取，而非固定常量；NISAR 为 L 波段，且 frequency A/B 的中心频率不同。显式设置 wavelength 可覆盖。
apply_water_mask 默认为 False —— ISCE3_NISAR 不执行 dem 阶段，没有可用的掩膜。
不提供 los_projection —— 'vertical' 需要处理器的 static 阶段，而 ISCE3_NISAR 不执行该阶段。

nisar_frequency / nisar_polarization 必须与处理器进行相位链接时所用的设置一致。

旧名称

ISCE3_Dolphin_PL_NISAR 仍解析到本分析器；ISCE3_Dolphin_PL_NISAR_Config 是 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.")

使用方法

创建分析器

from insarhub import Analyzer

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

或使用显式配置：

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)

运行

对该 workdir 中的堆叠运行 dolphin 时间序列反演（每个日期的累计位移、速度、残差）。
```
analyzer.run()
```