处理器

InSARHub 处理器模块专门提供干涉图处理功能。

导入处理器

导入 Processor 类以访问所有处理器功能
```
from insarhub import Processor
```
查看可用处理器

列出所有已注册的处理器
```
Processor.available()
```

可用处理器

Hyp3_S1ISCE2_S1GMTSAR_S1ISCE3_BurstISCE3_NISAR

HyP3 InSAR 处理器是 ASF HyP3 系统提供的基于云端的处理服务，用于从 Sentinel-1 SAR 数据生成干涉图。 InSARHub 将 hyp3_sdk 封装为其处理后端之一。

Hyp3_S1 专门封装了 hyp3_sdk 中的 insar_job，提供 InSAR SLC 处理工作流。

Source code in src/insarhub/processor/hyp3_s1.py

class Hyp3_S1(Hyp3Base):
    name = "Hyp3_S1"
    description = "HyP3 InSAR GAMMA processing. Produces geocoded interferograms from Sentinel-1 SLC pairs."
    compatible_downloader = "S1_SLC"
    default_config = Hyp3_S1_Config
    def __init__(self, config: Hyp3_S1_Config | None = None):
        super().__init__(config)
        # Fetch InSAR specific cost table
        try:
            self.cost = self.client.costs()['INSAR_GAMMA']['cost_table'][f'{self.config.looks}']
        except Exception as e:
            print(f"{Fore.YELLOW}Warning: Could not fetch InSAR cost table ({e}). Using local cost table.{Fore.RESET}")
            if self.config.looks == "20x4":
                self.cost = 10
            elif self.config.looks == "10x2":
                self.cost = 15
            else:
                raise ValueError(f"{Fore.RED}Unsupported looks configuration: {self.config.looks}. Please provide a valid cost for this looks setting.{Fore.RESET}")

    def submit(self):
        """
        Submit InSAR jobs to HyP3 based on the current configuration.

        Prepares job payloads from the `pairs` in the configuration
        and submits them via `_submit_job_queue`, handling user rotation,
        batching, and credit checks.

        The job names are automatically generated using the `name_prefix`
        and scene IDs.

        Raises:
            ValueError: If `self.config.pairs` is not a tuple of two strings
                        or a list of tuples of two strings.

        Returns:
            dict:
                A dictionary mapping usernames to lists of submitted `Batch` objects.

        Example:
            ```python
            processor = Hyp3InSARProcessor(config)
            batches = processor.submit()
            for user, batch in batches.items():
                print(f"{user} submitted {len(batch)} jobs")
            ```
        """

        # Normalize pairs input
        if isinstance(self.config.pairs, (list, tuple)) and all(isinstance(p, str) for p in self.config.pairs):
                pairs = [(self.config.pairs[0], self.config.pairs[1])]
        elif isinstance(self.config.pairs, (list, tuple)) and all(isinstance(p, tuple) for p in self.config.pairs):
            pairs = self.config.pairs
        else:
            raise ValueError(f"{Fore.RED}Invalid pairs format. Provide a list of tuples or a tuple of two strings.\n")

        job_queue: list[dict] = []

        for (ref_id, sec_id) in pairs:
            # We use the client to help format the dict, but we don't submit yet.
            # We are preparing the payload for _submit_job_queue
            job = self.client.prepare_insar_job(
                granule1=ref_id,
                granule2=sec_id,
                name= f"{self.config.name_prefix}_{ref_id.split('_')[5]}_{sec_id.split('_')[5]}",
                include_look_vectors=self.config.include_look_vectors,
                include_inc_map = self.config.include_inc_map,
                looks = self.config.looks,
                include_dem=self.config.include_dem,
                include_wrapped_phase=self.config.include_wrapped_phase,
                apply_water_mask=self.config.apply_water_mask,
                include_displacement_maps=self.config.include_displacement_maps,
                phase_filter_parameter=self.config.phase_filter_parameter
            )
            job_queue.append(job)

        # Send to base class for batching and submission
        batchs = self._submit_job_queue(job_queue)
        self.batchs = batchs
        return batchs

使用方法

使用参数创建处理器

使用搜索条件初始化处理器实例

processor = Processor.create('Hyp3_S1', workdir='/your/work/path', pairs=pairs)

或

params = {
    "workdir": '/your/work/path',
    "pairs": pairs,
}
processor = Processor.create('Hyp3_S1', **params)

或

from insarhub.config.defaultconfig import Hyp3_S1_Config
cfg = Hyp3_S1_Config(workdir='/your/work/path', pairs=pairs)
processor = Processor.create('Hyp3_S1', config=cfg)

Attributes:

Name	Type	Description
`workdir`	`Path \| str`	Directory where downloaded products will be stored. If provided as a string, it will be converted to a resolved `Path` object during initialization.
`saved_job_path`	`Path \| str \| None`	Optional path to a saved job JSON file for reloading previously submitted jobs. If provided as a string, it will be converted to a resolved `Path` object.
`earthdata_credentials_pool`	`dict[str, str] \| None`	Dictionary mapping usernames to passwords for managing multiple Earthdata accounts. Used for parallel or quota-aware submissions.
`skip_existing`	`bool`	If True, skip submission or download of products that already exist locally.
`submission_chunk_size`	`int`	Number of jobs submitted per batch request to the API. Helps avoid request size limits and API throttling.
`max_workers`	`int`	Maximum number of worker threads used for concurrent submissions or downloads. Recommended to keep below 8 to avoid overwhelming the API or triggering rate limits.

Attributes:

Name	Type	Description
`pairs`	`list[tuple[str, str]] \| None`	List of Sentinel-1 scene ID pairs in the form [(reference_scene, secondary_scene), ...]. If None, pairs must be provided during submission.
`name_prefix`	`str \| None`	Prefix added to generated HyP3 job names.
`include_look_vectors`	`bool`	If True, include look vector layers in the output product.
`include_los_displacement`	`bool`	If True, include line-of-sight (LOS) displacement maps.
`include_inc_map`	`bool`	If True, include incidence angle maps.
`looks`	`str`	Multi-looking factor in the format "range x azimuth" (e.g., "20x4").
`include_dem`	`bool`	If True, include the DEM used during processing.
`include_wrapped_phase`	`bool`	If True, include wrapped interferometric phase output.
`apply_water_mask`	`bool`	If True, apply a water mask during processing.
`include_displacement_maps`	`bool`	If True, include unwrapped displacement maps.
`phase_filter_parameter`	`float`	Phase filtering strength parameter (typically between 0 and 1). Higher values apply stronger filtering.

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

@dataclass
class Hyp3_S1_Config(Hyp3_Base_Config):
    """
    Configuration options for `hyp3_sdk` InSAR GAMMA processing jobs.

    This dataclass defines all parameters used when submitting
    InSAR jobs to the ASF HyP3 service using the GAMMA workflow.

    UI metadata is stored in ``_ui_groups`` / ``_ui_fields`` and consumed
    by the API layer to auto-generate the settings panel.

    Attributes:
        pairs (list[tuple[str, str]] | None):
            List of Sentinel-1 scene ID pairs in the form
            [(reference_scene, secondary_scene), ...].
            If None, pairs must be provided during submission.

        name_prefix (str | None):
            Prefix added to generated HyP3 job names.

        include_look_vectors (bool):
            If True, include look vector layers in the output product.

        include_los_displacement (bool):
            If True, include line-of-sight (LOS) displacement maps.

        include_inc_map (bool):
            If True, include incidence angle maps.

        looks (str):
            Multi-looking factor in the format "range x azimuth"
            (e.g., "20x4").

        include_dem (bool):
            If True, include the DEM used during processing.

        include_wrapped_phase (bool):
            If True, include wrapped interferometric phase output.

        apply_water_mask (bool):
            If True, apply a water mask during processing.

        include_displacement_maps (bool):
            If True, include unwrapped displacement maps.

        phase_filter_parameter (float):
            Phase filtering strength parameter (typically between 0 and 1).
            Higher values apply stronger filtering.
    """

    # ── UI metadata consumed by the API / settings panel ─────────────────────
    _ui_groups: ClassVar[list] = [
        {"label": "Processing",
         "fields": ["looks", "phase_filter_parameter", "name_prefix", "apply_water_mask"]},
        {"label": "Outputs",
         "fields": ["include_dem", "include_look_vectors", "include_inc_map",
                    "include_los_displacement", "include_wrapped_phase", "include_displacement_maps"]},
        {"label": "Job",
         "fields": ["skip_existing", "submission_chunk_size", "max_workers"]},
    ]
    _ui_fields: ClassVar[dict] = {
        "looks":                    {"type": "select", "options": ["20x4", "10x2"],
                                     "hint": "Range × azimuth looks (20x4 ≈ 80 m, 10x2 ≈ 40 m)"},
        "phase_filter_parameter":   {"type": "number", "min": 0, "max": 1, "step": 0.1,
                                     "default": 0.6,
                                     "hint": "Goldstein filter strength (0 = off, 1 = maximum)"},
        "name_prefix":              {"type": "text"},
        "apply_water_mask":         {"type": "bool"},
        "include_dem":              {"type": "bool"},
        "include_look_vectors":     {"type": "bool"},
        "include_inc_map":          {"type": "bool"},
        "include_los_displacement": {"type": "bool"},
        "include_wrapped_phase":    {"type": "bool"},
        "include_displacement_maps":{"type": "bool"},
        "skip_existing":            {"type": "bool",
                                     "hint": "Skip re-downloading already-completed jobs"},
        "submission_chunk_size":    {"type": "number", "min": 1, "max": 500, "step": 1,
                                     "default": 200,
                                     "hint": "Jobs per API batch request"},
        "max_workers":              {"type": "number", "min": 1, "max": 16, "step": 1,
                                     "default": 4,
                                     "hint": "Parallel download threads for completed job outputs (default 4)"},
    }
    # ─────────────────────────────────────────────────────────────────────────

    name: str = "Hyp3_S1_Config"
    pairs: list[tuple[str, str]] | None = None
    name_prefix: str | None = 'ifg'
    include_look_vectors:bool=True
    include_los_displacement:bool=False
    include_inc_map:bool=True
    looks:str='20x4'
    include_dem :bool=True
    include_wrapped_phase :bool=False
    apply_water_mask :bool=True
    include_displacement_maps:bool=True
    # 0.5 aligns the Goldstein filter strength with ISCE2_S1.filter_strength
    # and GMTSAR's phasefilt (hardcoded alpha=0.5), so the three backends
    # are comparable. ASF's own HyP3 default is 0.6.
    phase_filter_parameter :float=0.5

提交任务

根据当前配置向 HyP3 提交 InSAR 任务。

jobs = processor.submit()

Raises:

Type	Description
`ValueError`	If `self.config.pairs` is not a tuple of two strings or a list of tuples of two strings.

刷新任务

刷新所有任务的状态。

jobs = processor.refresh()

Raises:

Type	Description
`ValueError`	If no jobs are loaded in memory.

重试失败任务

通过重新提交来重试所有失败的任务。
```
jobs = processor.retry()
```
下载成功任务

下载所有用户的已成功任务。
```
processor.download()
```

保存当前任务

将当前任务批次信息保存到 JSON 文件。

processor.save()

Parameters:

Name	Type	Description	Default
`save_path`	`Path \| str \| None`	Path to save the batch JSON file. If None, defaults to `hyp3_jobs.json` in `self.output_dir`.	`None`

Raises:

Type	Description
`ValueError`	If no job batches exist to save.

监控任务

持续监控任务并下载已完成的输出。

processor.watch()

Parameters:

Name	Type	Description	Default
`refresh_interval`	`int`	Time interval (in seconds) between refreshes.	`300`

加载已保存任务

加载之前保存的 JSON 文件并恢复工作。
```
processor = Processor.create('Hyp3_S1', saved_job_path='path/to/your/json/file.json')
```
加载后可恢复检查/下载提交至 HyP3 服务器的任务。

ISCE2_S1 处理器在本地运行 ISCE2 stackSentinel，从下载的 SLC .SAFE 文件生成 Sentinel-1 干涉图。它生成一系列编号运行脚本并顺序执行，在每个步骤内并行运行独立命令。

导入处理器
```
from insarhub import Processor
```

创建处理器

from insarhub.config import ISCE2_S1_Config

cfg = ISCE2_S1_Config(
    workdir='/data/p100_f466',
    bbox=[33.0, 38.0, -120.0, -115.0],   # [南, 北, 西, 东]
)
pairs = [('20200101', '20200113'), ('20200113', '20200125')]
processor = Processor.create('ISCE2_S1', pairs=pairs, config=cfg)

Attributes:

Name	Type	Description
`workdir`	`Path \| str`	Processing root. All outputs (run_files/, merged/, etc.) live here.
`slc_dir`	`Path \| str`	Directory containing all Sentinel-1 SLC .SAFE files (or .zips).
`orbit_dir`	`Path \| str \| None`	Directory with .EOF orbit files. Created automatically if absent.
`aux_dir`	`Path \| str \| None`	Directory with Sentinel-1 AUX_CAL files. Defaults to workdir/aux; ISCE2 downloads missing files there on first run.
`dem_path`	`Path \| str \| None`	ISCE2-binary DEM (dem.wgs84 + .xml sidecar). When None, GLO-30 is pre-downloaded, preferring the joint footprint of the actual SLCs found in slc_dir/workdir (union of every scene's manifest corners, covering every frame in a merged multi-frame stack) over bbox -- the search AOI only reflects what was searched for, not what ASF actually returned (whole-scene footprints extend beyond it) or what a merge combined. bbox is used only as a fallback when no SLCs are on disk yet to derive a footprint from.
`isce_home`	`Path \| str \| None`	ISCE2 installation root. Falls back to $ISCE_HOME env var.
`bbox`	`list[float] \| None`	Area of interest as [S, N, W, E] degrees. Only used as a DEM bbox fallback when no SLCs are present yet to auto-derive a footprint from (see dem_path above); otherwise informational.
`num_overlap_connections`	`int`	Connections used for NESD azimuth coregistration.
`reference_date`	`str \| None`	Stack reference date YYYYMMDD. None = stackSentinel auto-selects.
`coregistration`	`str`	'NESD' (default, more accurate) or 'geometry' (faster).
`max_workers`	`int`	Parallel commands within each run step. This is the knob that controls concurrency for every step except topo -- InSARHub runs each run-file line itself under a ThreadPoolExecutor.
`num_proc4topo`	`int`	ISCE2's own multiprocessing pool size for the topo step (run_01), written into config_reference as `numProcess`. run_01 is a single command, so max_workers cannot parallelise it and this is the only knob that will. On HPC it is overridden by run_01's `sbatch_options.json` cpus_per_task.
`num_proc`	`int`	No effect under InSARHub. Hidden from the GUI for that reason; kept as a field so saved configs round-trip and because `_resolve_num_proc()` still reads it as the HPC fallback. stackSentinel's `--num_proc` does exactly one thing: decide which lines of a run file get a trailing `&` and where `wait` goes (Stack.py::write_wrapper_config2run_file), i.e. shell-level parallelism. InSARHub strips those `&` in `_fix_cmd` -- it has to, since `subprocess.run("cmd &", shell=True)` returns instantly with rc=0 and reports orphaned work as success -- and schedules the commands under its own pool instead. So the value never reaches ISCE2 as a process count: of the 51 configs stackSentinel writes for a 4-scene stack, only config_reference carries `numProcess`, and that one comes from num_proc4topo. Set max_workers instead.

提交（本地模式）

生成运行脚本并在后台进程中开始顺序执行。立即返回；使用 refresh() 监控进度。
```
jobs = processor.submit()
```
提交（HPC / SLURM 模式）

设置 hpc_mode=True 启用滑动窗口 SLURM 管理器。步骤会先分组：每场景/每对命令数相同的连续步骤会合并为单个组管理器（例如 run_02_unpack_secondary_slc 与 run_03_average_baseline 若每个场景各有一条命令，就会合并）；其余步骤各自拥有独立的单步管理器。每个管理器随时保持最多 max_concurrent_hpc 个子作业同时运行，有空槽时立即补充。每个 sbatch 脚本按命令记录带耗时秒数的 START/DONE/FAIL 日志。

submit() 只直接提交第一个分组的管理器。此后每个管理器在自身成功完成后，会通过自己脚本末尾的 sbatch 调用去提交下一个分组的管理器——而不是依赖 SLURM 的 --dependency——因此任意时刻队列中最多只会有一个管理器（加上它自己不超过 max_concurrent_hpc 个的子作业），而不是把所有分组的管理器一次性提前全部提交。这一点很关键，因为 SLURM 按用户限制提交作业数的 QOS 上限，对"仅仅在等待依赖"的作业和正在运行的作业是一视同仁地计数的；一次性提前提交整条链，可能会让本来什么都没做、只是在排队等轮到自己的管理器把这个上限占满。若某个管理器失败或被取消，它就不会再提交下一个，链条自然中断——不需要额外清理尚未提交的剩余部分。refresh() 会自动读取新链式提交作业的 ID（管理器会把它写入该分组日志目录旁的小文件 chained_job_id.txt）。

管理器作业名简短且能直接看出它管理哪个/哪些运行步骤：单步管理器为 i<NN>_mgr（例如 run_04_... 对应 i04_mgr），组管理器为跨越步骤 NN–MM 的 i<NN>-<MM>_grp（例如 i02-03_grp）——便于在 squeue 中一眼辨认。
```
cfg = ISCE2_S1_Config(
    workdir='/data/p100_f466',
    bbox=[33.0, 38.0, -120.0, -115.0],
    hpc_mode=True,
    max_concurrent_hpc=12,   # 默认值；根据集群公平份额限制调整
)
processor = Processor.create('ISCE2_S1', pairs=pairs, config=cfg)
processor.submit()
```
retry() 从已保存的作业元数据（slurm_job_ids / hpc_manager / hpc_array）自动检测 HPC 模式，无需再次传入 hpc_mode=True。

试运行

预览运行脚本和路径检查，不执行任何操作。

cfg = ISCE2_S1_Config(
    workdir='/data/p100_f466',
    bbox=[33.0, 38.0, -120.0, -115.0],
    dry_run=True,
)
processor = Processor.create('ISCE2_S1', pairs=pairs, config=cfg)
processor.submit()

刷新

从磁盘读取步骤和命令状态。
```
jobs = processor.refresh()
```
重试失败步骤

重新运行所有状态为 FAILED 的步骤。
```
processor.retry()
```
取消

终止正在运行的后台进程（本地模式）或对所有活动 SLURM 任务执行 scancel（HPC 模式）。
```
processor.cancel()
```
监控

定期轮询步骤状态，直到所有步骤完成。
```
processor.watch(refresh_interval=60)
```

保存 / 加载

任务状态在 submit() 后自动保存。从已保存的任务文件重新加载并恢复：

cfg = ISCE2_S1_Config(
    workdir='/data/p100_f466',
    saved_job_path='/data/p100_f466/isce/isce_jobs_<timestamp>.json',
)
processor = Processor.create('ISCE2_S1', pairs=[], config=cfg)
processor.refresh()   # 或 .retry()、.cancel()、.watch()

无需本地安装 ISCE2

将 container 字段设置为 Apptainer/Singularity .sif 镜像的路径，或 Docker 镜像引用（name[:tag]），submit()/retry()/refresh()/watch()/cancel() 都会在容器内而非宿主机上重新执行同一个 insarhub processor ... CLI 调用 — 工作目录会以相同路径绑定挂载，因此输出会像本机运行一样落在原处，ISCE2 也完全不需要在宿主机上被发现。容器镜像只需在 ISCE2/topsStack 旁额外安装 insarhub（可参考 docker/dev/Dockerfile.isce2-mintpy 作为现成示例）。运行应用/CLI 的宿主机需要在 PATH 上有容器运行时（docker，或对 .sif 用 apptainer/singularity）；容器镜像本身不需要 — 流水线直接在镜像内运行，不会再嵌套一次 docker run。
```
cfg = ISCE2_S1_Config(
    workdir='/data/p100_f466',
    bbox=[33.0, 38.0, -120.0, -115.0],
    container='ghcr.io/jldz9/insarhub-isce2-mintpy:0.4.0',
)
processor = Processor.create('ISCE2_S1', pairs=pairs, config=cfg)
processor.submit()
```
CLI 用法相同：
```
insarhub processor -N ISCE2_S1 -w /data/p100_f466 submit \\
    --container ghcr.io/jldz9/insarhub-isce2-mintpy:0.4.0
```
container 会持久化写入工作目录的 insarhub_config.json，因此之后的 retry()/refresh()/cancel()（以及 GUI 中的重试）会在同一镜像内重新运行，无需再次传入。后续调用若显式给出 --container / container= 会覆盖已保存的值；不带值的 --container 会解析为处理器的 container_default 镜像。container_default 是每个处理器固定的建议镜像（GUI “在容器中运行” 复选框会用它预填），从不持久化 —— 只有你实际选择的 container 会被保存。在 HPC 模式下，只有各阶段的子作业在容器内运行，sbatch 管理器脚手架仍留在宿主机上。

在本地运行 GMTSAR 的 Python 流程，从 .SAFE SLC 生成 Sentinel-1 干涉图。入口由 subswath 决定：

单个 IW（例如 2）——单子条带，走 p2p_processing
多个 IW（例如默认的 "1 2 3"）——多子条带合并，走 p2p_S1_TOPS_Frame

两种模式下调用方都只需传入原始的 .SAFE/.EOF 名称，子条带与极化方式由内部提取。

GMTSAR 运行在自己的 conda 环境中。gmtsar_root 与 gmtsar_env_bin 用于定位它，两者均可自动探测，仅在探测失败时才需显式指定。也可将 container 设为包含 insarhub+GMTSAR 的 .sif/Docker 镜像，从而跳过本地探测。

导入处理器
```
from insarhub import Processor
```

创建处理器

from insarhub.config import GMTSAR_S1_Config

cfg = GMTSAR_S1_Config(
    workdir       = '/data/stack',
    slc_dir       = '/data/slcs',
    orbit_dir     = '/data/orbits',
    dem_path      = '/data/dem.grd',   # GMTSAR 格式 DEM；未设置时在暂存阶段自动下载
    subswath      = 2,                 # 仅 IW2 — 单子条带。默认 "1 2 3" 为多子条带合并
)
pairs = [
    ("REF.SAFE", "REF.EOF", "SEC.SAFE", "SEC.EOF"),
]
processor = Processor.create('GMTSAR_S1', pairs=pairs, config=cfg)

多子条带模式下请设置 dem_path

多子条带模式为每个干涉对建立独立的 case 目录。若 dem_path 未设置，DEM 会在暂存阶段自动下载 — 也就是每个干涉对下载一次。对 27 个干涉对的网络而言，这意味着重复下载 27 次同一个 DEM。

Attributes:

Name	Type	Description
`workdir`	`Path \| str`	Processing root. gmtsar/ (raw/, topo/, config.py, intf/ or per-pair subdirs) lives here.
`slc_dir`	`Path \| str \| None`	Directory containing Sentinel-1 SLC .SAFE dirs (or .zips).
`orbit_dir`	`Path \| str \| None`	Directory with .EOF orbit files.
`dem_path`	`Path \| str \| None`	GMTSAR-format DEM grid (topo/dem.grd). Unlike ISCE2_S1, bbox-driven auto-download is NOT implemented yet -- must be supplied explicitly. See gmtsar_s1.py's module docstring for the concrete "known gaps" list.
`sat`	`str`	p2p_processing's SAT argument (single-subswath mode only). Exposed (not hardcoded) for forward-compat -- GMTSAR already supports 14 sensor families beyond S1_TOPS (see gmtsar/python/tests/cases.py upstream), this processor is just the first one wired in.
`subswath`	`int \| str`	IW subswath(s), ISCE-style space-separated (e.g. "1 2 3" = full frame merged via p2p_S1_TOPS_Frame, "2" = single-subswath via p2p_processing). p2p_processing itself does not read .SAFE directories -- it expects one subswath's .tiff/.xml files already extracted to matching-stem files in raw/ (confirmed against GMTSAR's own bundled single-subswath test fixture, H_res/raw/: its per-stem files are plain symlinks into the equivalent Frame-mode F/ subswath files pulled from the same .SAFE). GMTSAR_S1 does this extraction itself. Default "1 2 3" (full frame, matching stack_mode's own default).
`parallel`	`bool`	p2p_S1_TOPS_Frame's own internal subswath parallelism flag (0=sequential, 1=parallel). Only used in multi-subswath mode.
`config_template`	`Path \| str \| None`	Path to a GMTSAR config.py to reuse as-is. If None, one is auto-generated per case via `pop_config <sat>` (GMTSAR's own default-config tool), matching p2p_processing's own "no config.py given" behavior.
`max_workers`	`int`	Independent pairs processed concurrently.
`skip_existing`	`bool`	Don't redo a pair whose output dir already has a .succeeded status marker.
`gmtsar_root`	`Path \| str \| None`	GMTSAR repo root ($GMTSAR). Required -- GMTSAR_S1 raises at construction time if unset. Its bin/ is prepended to every GMTSAR subprocess call's PATH.
`gmtsar_env_bin`	`Path \| str \| None`	bin/ dir of the conda env GMTSAR needs (provides the real `gmt` binary plus numba/scipy). Required -- InSARHub's own env does not provide `gmt` at all (confirmed via a real end-to-end test, 2026-07-21), so subprocess calls fail near-instantly without this. See gmtsar_s1.py's _subprocess_env() docstring for the full writeup.

提交

暂存 GMTSAR case 目录（单子条带模式下还会提取每个干涉对的子条带），随后在后台启动 p2p_processing/p2p_S1_TOPS_Frame，最多同时运行 max_workers 个干涉对。该调用立即返回；请用 refresh()/watch() 查看进度。
```
jobs = processor.submit()
```
提交（HPC / SLURM 模式）

两种模式都支持 hpc_mode=True。

p2p 模式（stack_mode=False，默认）是两者中较简单的：干涉对之间完全独立 — 多子条带模式下各自拥有独立的 case 目录，单子条带模式的输出以 intf/<julian_pair>/ 隔离 — 因此由单个滑动窗口管理器一次性展开全部干涉对，同时最多 max_concurrent_hpc 个，完全不需要链式提交。每个子作业通过内部的 run-stage-unit --stage pair --index N 重入，运行该干涉对的完整流程（配准 → 干涉图 → 滤波 → 解缠 → 地理编码）。作业名为 g_p2p_mgr / g_p2p_<idx>。
```
cfg = GMTSAR_S1_Config(workdir='/data/stack', hpc_mode=True)
Processor.create('GMTSAR_S1', pairs=pairs, config=cfg).submit()
```
stack_mode 则把每个栈阶段（align_F<N>/intf_F<N>/merge，单子条带时为扁平的 align/intf）作为各自的滑动窗口管理器运行，与 ISCE2_S1 采用相同的链式提交设计：只直接提交第一个阶段的管理器，其余每个管理器在自己成功后再提交下一个 — 而不是预先排入 --dependency 依赖链 — 因此队列中同时最多只有一个管理器（外加它自己的 ≤max_concurrent_hpc 个子作业）。

与 ISCE2_S1 的一个真实差异：GMTSAR_S1 没有 stackSentinel.py 的 run_NN_* 那种扁平 shell 命令清单 — 每个阶段的实际工作都在 Python 方法中（_run_align_unit/_run_intf_unit/_run_merge_unit），因此每个 HPC 子作业的"命令"都是重新进入 insarhub 自身（内部的 run-stage-unit CLI 动作），在新进程中调用其中一个方法，而不是直接执行调用 GMTSAR 可执行文件的 shell 命令行。

详见 HPC（SLURM）。
刷新状态

从 GMTSAR 自身的输出标记读取每个干涉对的状态。p2p 模式下会以每个干涉对一行的彩色表格显示，并用 SLURM 的实时状态覆盖过期的标记文件。
```
jobs = processor.refresh()
```
重试失败的干涉对

只重新运行状态为 FAILED 的干涉对。
```
processor.retry()
```
等待完成

轮询直到每个干涉对都为 SUCCEEDED 或 FAILED。
```
processor.watch()
```
保存状态

submit() 之后作业状态会自动保存到 <workdir>/gmtsar/gmtsar_jobs.json。
```
processor.save()
```
取消（HPC 模式）

对 HPC 提交执行 scancel，涵盖管理器与全部子作业，两种模式均适用。p2p 的作业是通过 hpc/p2p/ 目录识别的，而不是依赖 config.hpc_mode，因此直接执行 cancel 即可找到它们、无需重复加 --hpc-mode；任何仍处于 PENDING/RUNNING 的干涉对会被标记为 FAILED，以免 refresh 继续把它显示为运行中。
```
processor.cancel()
```
输出目录结构

单子条带：<workdir>/gmtsar/intf/<julian_date_pair>/（例如 intf/2019184_2019196/ — 这是 GMTSAR 自己的儒略日命名，而不是 ref/sec 词干）— 保持 GMTSAR 的原生文件名（corr_ll.grd、phasefilt_ll.grd、*.PRM），正是 MintPy 的 prep_gmtsar.py 直接期望的形式。

多子条带：<workdir>/gmtsar/<ref_safe>_<sec_safe>/merge/ — 跨所有指定子条带的合并、地理编码产品（phasefilt_ll.grd、corr_ll.grd，以及 PNG/KML 预览图）。
时序分析：请使用 MintPy，而非 GMTSAR 自带的 sbas

这是选择 p2p 的直接后果。GMTSAR 自带的 sbas 在雷达坐标下运行，要求所有 SLC 都重采样到同一个公共网格 — 而 p2p 的逐对配准并不提供这一点。MintPy 的 prep_gmtsar 读取的是地理编码后的 *_ll.grd，所有干涉对本来就在同一地理网格上，因此无需公共配准主影像。请使用 GMTSAR_Mintpy_SBAS 分析器。
无需本地安装 GMTSAR 即可运行

与 ISCE2_S1 相同，把 container 设为一个装有 insarhub + GMTSAR 的 .sif 或 Docker 镜像，即可跳过本地探测：
```
insarhub processor -N GMTSAR_S1 -w /data/stack submit \
    --container ghcr.io/jldz9/insarhub-gmtsar-mintpy:0.4.0
```
在 HPC 模式下，只有各阶段的子作业在容器内运行，sbatch 管理器脚手架仍留在宿主机上。

从 ASF SLC-BURST 数据构建干涉图栈，使用 ISCE3/COMPASS 进行地理编码，其后环节由 dolphin 完成。请与 S1_Burst 下载器配合使用。

不做配准：COMPASS 把每一景独立地理编码到绝对 UTM 坐标，因此同一 burst 的任意两个日期在构造上即逐像元对齐。

共九个阶段，按顺序执行：

阶段	工具	产出
`dem`	`sardem`	Copernicus DEM + NASADEM 水体掩膜
`tec`	COMPASS	每个采集日期一个 IONEX 电离层图
`cslc`	`s1_geocode_stack.py` → `run_*.sh`	每个 burst-日期一个地理编码 CSLC
`static`	`s1_static_layers.py`	LOS/入射角几何量，随后重采样到栈网格
`crop`	dolphin	按 AOI 裁剪每个 burst
`ifg`	dolphin	干涉图（见 `ifg_mode`）
`stitch`	dolphin	把每个干涉对的各 burst 拼接成一景
`filt`	dolphin	多视 → Goldstein 滤波 → 相干性
`unwrap`	snaphu	解缠相位 + 连通分量

选择估计方式 — `ifg_mode`

取值	干涉对来源	说明
`phase_link`（默认）	完整协方差估计	所有干涉对都参与；由 `pl_*` 参数调节
`network`	规则生成	`n_connections` 或 `max_temporal_baseline`
`user_defined`	本目录的 `stack_*.json`	恰好使用 `select_pairs` 选出的干涉对

默认使用 phase_link，因为它的效果有实测优势：在测试栈上得到 1 个连通分量、覆盖率 83%，而逐对网络为 3 个、55%；闭合误差从 0.157 rad 降到 0.067 rad。其参数与 dolphin 自身发布的配置一致（glrt / 0.001、半窗 7×14、ministack 15）。

在 phase_link + pl_ifg_network=single_reference（均为默认值）下，用户自定义网络会被忽略 — 估计器的输出本身就是那个网络。若希望在该模式下仍使用自己的干涉对，请设置 pl_ifg_network=bandwidth。

处理范围

AOI 会自动取自本目录下载器配置中的 intersectsWith，通常已经填好。若要改为处理整个已下载的 burst 范围，勾选 process_full_extent。注意 crop_buffer_deg（默认 0.05°）会在四周各外扩一圈 — 对较小的 AOI 而言，这个缓冲本身就可能接近整个 burst 范围，因此若希望 AOI 真正起到裁剪作用，请调小该值。

dem 与 cslc 在任何地理编码完成之前运行，因此始终使用 AOI；process_full_extent 从 crop 阶段起才生效。

其他说明

各阶段针对 SLURM 做了拆分 — 参见 HPC（SLURM）。cslc 是每个 burst-日期一个作业，也是耗时主体；phase_link 下的 ifg 是单个作业，因为估计器没有逐干涉对的单元。
干涉图网络取各 burst 日期列表的交集。若 ASF 在某天缺少某个 burst 的数据，该日期会被剔除并明确列出，从而保证每个干涉对在所有 burst 上都能生成。
时序分析使用 ISCE3_Dolphin_S1_PL 分析器，它同时支持两种估计方式。

Source code in src/insarhub/processor/isce3_burst.py

class ISCE3_Burst(ISCE3_Base):
    name = "ISCE3_Burst"
    description = ("Sentinel-1 burst InSAR via ISCE3 + COMPASS: geocoded burst "
                   "SLCs (no coregistration), stitched interferograms, snaphu.")
    default_config = ISCE3_Burst_Config
    compatible_downloader = "S1_Burst"

    #: job-file name; kept burst-specific so a workdir running another ISCE3
    #: processor keeps the two job files apart.
    JOBS_FILE = "isce3_burst_jobs.json"

    #: stage order. The dolphin engine (ifg/stitch/unwrap) delegates each stage
    #: to the exact function dolphin's own displacement.run calls, so output is
    #: byte-identical to `dolphin run` by construction.
    #: `los` is LAST: it needs the stitched interferogram grid to resample onto.
    STAGES = ("dem", "tec", "cslc", "static", "ifg", "stitch", "unwrap", "los")
    _IMPLEMENTED = ("dem", "tec", "cslc", "static", "ifg", "stitch", "unwrap", "los")

    # dolphin drives ifg..unwrap; compass drives cslc/static/tec. Both are
    # imported inside the stage methods, so a child job only discovers a
    # missing one after it starts -- hence the pre-submit check.
    REQUIRED_MODULES = ("dolphin", "compass")

    # Per-stage SLURM resources for a fresh workdir. Sized by what each stage
    # actually does: cslc runs COMPASS geocoding per burst-date (the long pole),
    # unwrap runs snaphu, dem/tec are network-bound and want a core and time
    # rather than memory.
    SBATCH_DEFAULT_TEMPLATE = {
        **sbatch_template_header(),
        "_stages": {
            "dem":    "Copernicus DEM download (sardem) + water mask",
            "tec":    "IONEX ionosphere maps, one per acquisition date",
            "cslc":   "COMPASS geocoding -- ONE JOB PER BURST-DATE, the long pole",
            "static": "static geometry layers, one job per burst, then LOS mosaic",
            "ifg":    "PS + phase-link + interferograms, ONE JOB PER BURST "
                      "(dolphin wrapped_phase.run)",
            "stitch": "mosaic bursts + estimate correlation, ONE JOB "
                      "(dolphin stitching_bursts.run)",
            "unwrap": "snaphu, ONE JOB (dolphin unwrapping.run)",
            "los":    "LOS/incidence geometry onto the stack grid -- runs LAST",
        },
        # "default" is the BASE every stage inherits (see
        # ISCE3_Base._stage_slurm_kwargs), so site-wide settings -- partition
        # above all -- belong here once rather than repeated per stage.
        # No "account": not every cluster requires one, and an invalid account
        # makes sbatch reject the job outright.
        "default": {"time": "02:00:00", "cpus_per_task": 4, "mem": "16G",
                    "partition": "all"},
        # Managers are pure bookkeeping and must outlive every child they
        # supervise, so give them the longest-walltime partition available.
        # Only "partition" is read here -- sizing is fixed by slurm_manager.
        "manager": {"partition": "all"},
        "dem":     {"time": "04:00:00", "cpus_per_task": 1, "mem": "8G"},
        "tec":     {"time": "02:00:00", "cpus_per_task": 1, "mem": "2G"},
        "cslc":    {"time": "04:00:00", "cpus_per_task": 4, "mem": "32G"},
        "static":  {"time": "02:00:00", "cpus_per_task": 4, "mem": "16G"},
        # Sized for phase_link: holds a ministack's worth of SLC covariance in
        # memory rather than multiplying two rasters.
        "ifg":     {"time": "08:00:00", "cpus_per_task": 8, "mem": "64G"},
        "stitch":  {"time": "01:00:00", "cpus_per_task": 2, "mem": "16G"},
        "unwrap":  {"time": "04:00:00", "cpus_per_task": 4, "mem": "24G"},
        "los":     {"time": "01:00:00", "cpus_per_task": 4, "mem": "16G"},
    }

    # ------------------------------------------------------------------
    # paths
    # ------------------------------------------------------------------

    @property
    def slc_dir(self) -> Path:
        # lowercase 'slc': S1_Burst writes .SAFE and their .EOF orbits together
        # there, the same layout S1_SLC produces. The GUI/config exposes this as
        # `burst_path` (the assembled .SAFE stack); `slc_dir` is kept as a
        # back-compat alias that still wins if set explicitly.
        v = getattr(self.config, "slc_dir", None)
        if v:
            return Path(v).expanduser()
        return self._p("burst_path", "slc")

    @property
    def orbit_dir(self) -> Path:
        v = getattr(self.config, "orbit_dir", None)
        return Path(v).expanduser() if v else self.slc_dir

    @property
    def dem_path(self) -> Path:
        return self._p("dem_path", "dem/cop_dem.tif")

    @property
    def tec_dir(self) -> Path:
        return self._p("tec_dir", "tec")

    @property
    def cslc_dir(self) -> Path:
        return self._p("cslc_dir", "cslc")

    # dolphin-engine products, in dolphin's native layout (identical to what
    # `dolphin run` writes): per-burst dirs under <workdir>/<burst_id>/ from the
    # ifg stage, stitched products under interferograms/ + unwrapped/.
    @property
    def stitch_dir(self) -> Path:
        """Stitched interferograms + correlation + temporal coherence."""
        return self._p("stitch_dir", "interferograms")

    @property
    def unwrap_dir(self) -> Path:
        return self._p("unwrap_dir", "unwrapped")

    def burst_dirs(self) -> list[Path]:
        """Per-burst output directories written by the ifg stage."""
        if not self.workdir.is_dir():
            return []
        pat = re.compile(_BURST_RE)
        return sorted(d for d in self.workdir.iterdir()
                      if d.is_dir() and pat.match(d.name))

    def quality_file(self) -> Path | None:
        """Stitched temporal coherence (dolphin's quality raster), by glob."""
        for p in sorted(self.stitch_dir.glob("temporal_coherence*.tif")):
            return p
        return None

    @property
    def water_mask_path(self) -> Path:
        """ESA WorldCover water mask (GeoTIFF, 1=land/0=water) by ``run_dem``.

        dolphin's mask convention -- consumed directly by the unwrap stage and by
        ``displacement.run``'s ``mask_file``, both of which warp it onto the
        interferogram grid, so no ``.wbd``/sidecar handling is needed.
        """
        return self.dem_path.parent / "water_mask.tif"

    # ------------------------------------------------------------------
    # inputs
    # ------------------------------------------------------------------

    def acquisition_dates(self) -> list[str]:
        """Sorted YYYYMMDD from the .SAFE names in slc_dir.

        Read off disk rather than from ``pairs`` so the DEM and TEC stages work
        straight after download, before any pairing has been decided.
        """
        out = set()
        for s in sorted(self.slc_dir.glob("*.SAFE")):
            m = re.search(r"_(\d{8})T\d{6}_", s.name)
            if m:
                out.add(m.group(1))
        return sorted(out)

    def dem_bbox(self) -> tuple[float, float, float, float]:
        """DEM footprint: AOI grown by ``dem_buffer_deg`` and snapped outward.

        The buffer is not cosmetic. Geocoding reaches beyond the AOI -- a burst
        extends past the target area, and range-Doppler terrain correction needs
        DEM coverage wherever the radar looks. A DEM cropped to the AOI leaves
        edge bursts with no elevation and produces void output there. The stack
        notebook uses 2 degrees for the same reason.
        """
        # Goes through the same resolver as the interferogram stages so DEM and
        # processing extent can never disagree. The DEM stage normally runs
        # BEFORE any burst is geocoded, so the full-extent fallback has nothing
        # to measure -- in that case config.AOI or the downloader's
        # intersectsWith is the only available answer, and _aoi() says so.
        import math
        w, s, e, n = self._aoi()
        b = float(getattr(self.config, "dem_buffer_deg", 2.0))
        return (math.floor(w - b), math.floor(s - b),
                math.ceil(e + b), math.ceil(n + b))

    # ------------------------------------------------------------------
    # stage: dem
    # ------------------------------------------------------------------

    def run_dem(self, force: bool = False) -> bool:
        """Copernicus DEM via sardem, plus the NASADEM water-body mask.

        Reproduces the stack notebook's section 1.3::

            sardem --bbox W S E N --output-type float32 --output-format GTiff \\
                   --data-source COP -o <dem_path>
            ut.download_nasadem_water_mask(dem_wsen, dem_path.parent)
        """
        dem = self.dem_path
        dem.parent.mkdir(parents=True, exist_ok=True)
        w, s, e, n = self.dem_bbox()

        if dem.exists() and not force:
            print(f"[ISCE3_Burst] DEM already present: {dem} "
                  f"({dem.stat().st_size / 1e6:.0f} MB) -- use force=True to redo")
        else:
            cmd = ["sardem", "--bbox", str(w), str(s), str(e), str(n),
                   "--output-type", "float32", "--output-format", "GTiff",
                   "--data-source", str(getattr(self.config, "dem_source", "COP")),
                   "-o", str(dem)]
            print(f"[ISCE3_Burst] DEM bbox {w} {s} {e} {n} "
                  f"(AOI + {getattr(self.config, 'dem_buffer_deg', 2.0)} deg)")
            print(f"  $ {' '.join(cmd)}")
            p = subprocess.run(cmd, env=self._env(), text=True,
                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
            if p.returncode != 0 or not dem.exists():
                logger.error("ISCE3_Burst: sardem failed (rc=%s): %s",
                             p.returncode, (p.stdout or "")[-600:])
                return False
            print(f"  -> {dem}  ({dem.stat().st_size / 1e6:.0f} MB)")

        if getattr(self.config, "water_mask", True):
            wbd = self.water_mask_path
            if wbd.exists() and not force:
                print(f"[ISCE3_Burst] water mask already present: {wbd.name}")
            elif not self._download_water_mask((w, s, e, n), wbd):
                # Non-fatal: the mask only blanks water before unwrapping.
                logger.warning("ISCE3_Burst: water mask unavailable; unwrapping "
                               "will proceed without masking open water")
        return True

    def _download_water_mask(self, bbox, out_path: Path) -> bool:
        """ESA WorldCover water mask (1=land/0=water) to ``out_path``.

        sardem's old ``NASA_WATER`` source (SRTM Water Body Data on
        ``e4ftl01.cr.usgs.gov``) was migrated off that host and 404s for every
        tile, so it silently produced an all-land mask. ESA WorldCover is on
        open AWS (no Earthdata/account) and marks permanent water as class 80.
        """
        try:
            from insarhub.utils.tool import download_worldcover_water_mask
        except Exception as exc:                                 # noqa: BLE001
            logger.warning("ISCE3_Burst: no water-mask downloader available (%s)",
                           exc)
            return False
        try:
            return bool(download_worldcover_water_mask(bbox, out_path))
        except Exception as exc:                                 # noqa: BLE001
            logger.warning("ISCE3_Burst: water mask download failed: %s", exc)
            return False

    # ------------------------------------------------------------------
    # stage: tec
    # ------------------------------------------------------------------

    def run_tec(self, force: bool = False) -> bool:
        """IONEX global ionosphere maps, one per acquisition date.

        Reproduces the stack notebook's section 1.4. COMPASS consumes these as a
        timing correction during geocoding (``los_ionospheric_delay``), so they
        must exist before the cslc stage -- a missing TEC file silently drops
        the ionospheric term rather than failing.
        """
        # COMPASS's own download_ionex hits CDDIS unauthenticated, which now
        # requires Earthdata Login -- it silently saves a URS login page instead
        # of the map. Use InSARHub's Earthdata-authenticated resolver, which
        # reuses COMPASS's filename logic so the files still land where the cslc
        # runconfigs expect them. (Importing get_ionex_filename here also asserts
        # COMPASS is present, as before.)
        try:
            from insarhub.utils.ionex import (download_ionex_earthdata,
                                              have_earthdata_creds)
            from compass.utils.iono import get_ionex_filename  # noqa: F401
        except ImportError as exc:                               # noqa: BLE001
            logger.error("ISCE3_Burst: TEC download needs COMPASS "
                         "(compass.utils.iono) plus insarhub.utils.ionex (%s).", exc)
            return False

        if not have_earthdata_creds():
            logger.error(
                "ISCE3_Burst: no Earthdata credentials in ~/.netrc "
                "(machine urs.earthdata.nasa.gov). CDDIS IONEX downloads require "
                "Earthdata Login -- run the S1 downloader once to store them, or "
                "add the entry by hand. Skipping the TEC stage; bursts will be "
                "geocoded without an ionospheric correction.")
            return False

        dates = self.acquisition_dates()
        if not dates:
            logger.error("ISCE3_Burst: no .SAFE found in %s -- run the S1_Burst "
                         "downloader first", self.slc_dir)
            return False

        tec = self.tec_dir
        tec.mkdir(parents=True, exist_ok=True)
        code = str(getattr(self.config, "tec_sol_code", "jpl"))
        # Analysis centres to try when the preferred one has no map for a date.
        # CDDIS coverage is NOT uniform across centres: JPL's archive has a real
        # hole from 2023-08-11 to 2023-10-10 (spanning its jplgDDD0.YYi ->
        # JPL0OPSFIN_*.INX rename), where igs/cod/esa are missing too but UPC
        # has every day. Verified live against CDDIS.
        fallbacks = [c for c in
                     (str(x) for x in (getattr(self.config, "tec_fallback_codes", None)
                                       or ("igs", "cod", "esa", "upc")))
                     if c and c != code]
        print(f"[ISCE3_Burst] TEC ({code}) for {len(dates)} date(s) -> {tec}")

        have = _ionex_dates_on_disk(tec)
        ok, used_fallback, missing = 0, [], []
        for d in dates:
            if d in have and not force:
                ok += 1
                continue
            got = None
            for i, c in enumerate([code, *fallbacks]):
                try:
                    got = download_ionex_earthdata(d, str(tec), sol_code=c)
                except Exception:                                # noqa: BLE001
                    continue
                if i:
                    used_fallback.append((d, c))
                break
            if got:
                ok += 1
            else:
                missing.append(d)

        if used_fallback:
            print(f"[ISCE3_Burst] {len(used_fallback)} date(s) fell back to another "
                  f"IONEX centre: " +
                  ", ".join(f"{d}={c}" for d, c in used_fallback[:6]) +
                  (" ..." if len(used_fallback) > 6 else ""))
        print(f"[ISCE3_Burst] TEC: {ok}/{len(dates)} date(s)")

        if missing:
            # A missing IONEX map is NOT fatal. COMPASS drops the ionospheric
            # delay term for that acquisition and geocodes it anyway, so the
            # cost is a small phase offset on pairs touching these dates -- not
            # a broken stack. Failing all 99 dates because a handful of days are
            # absent from every public archive would be disproportionate, so
            # this warns loudly and continues.
            logger.warning(
                "ISCE3_Burst: no IONEX map from any centre for %d date(s): %s. "
                "Those acquisitions will be geocoded WITHOUT an ionospheric "
                "correction.", len(missing), ", ".join(missing))
            print(f"[ISCE3_Burst] WARNING: {len(missing)} date(s) have no TEC from "
                  f"any centre and will be geocoded without an ionospheric "
                  f"correction: {', '.join(missing[:8])}"
                  + (" ..." if len(missing) > 8 else ""))
        # TEC is non-fatal: the cslc stage geocodes WITHOUT the ionospheric
        # correction when no IONEX map exists (a small phase offset, not a
        # broken stack). Returning `ok > 0` here made an all-missing TEC stage
        # (e.g. CDDIS behind Earthdata auth) abort the whole pipeline.
        return True

    # ------------------------------------------------------------------
    # stage: cslc
    # ------------------------------------------------------------------

    @property
    def burst_db_path(self) -> Path:
        v = getattr(self.config, "burst_db_path", None)
        if v:
            return Path(v).expanduser()
        # notebook default: a sibling of work_dir, shared across projects
        return self.workdir.parent / "s1-burst-db" / "opera-burst-bbox-only.sqlite3"

    _BURST_DB_URL = ("https://github.com/opera-adt/burst_db/releases/download/"
                     "v0.10.0/opera-burst-bbox-only.sqlite3")

    def _ensure_burst_db(self) -> bool:
        db = self.burst_db_path
        if db.exists():
            return True
        db.parent.mkdir(parents=True, exist_ok=True)
        print(f"[ISCE3_Burst] fetching OPERA burst DB -> {db}")
        try:
            from urllib.request import urlretrieve
            urlretrieve(self._BURST_DB_URL, db)
        except Exception as exc:                                 # noqa: BLE001
            logger.error("ISCE3_Burst: burst DB download failed: %s", exc)
            return False
        return db.exists()

    def tec_map(self) -> dict[str, str]:
        """{YYYYMMDD: ionex_path} from the files the tec stage downloaded.

        Handles both IONEX naming conventions the notebook does: the long IGS
        product name (``JPL0OPSFIN_20242350000_01D_02H_GIM.INX``, where the date
        is year+day-of-year) and the legacy short form (``jplg2350.24i``).
        """
        from datetime import timedelta
        out: dict[str, str] = {}
        for f in sorted(self.tec_dir.glob("*GIM.INX")):
            m = re.search(r"_(\d{4})(\d{3})\d{4}_", f.name)
            if m:
                d = datetime(int(m.group(1)), 1, 1) + timedelta(days=int(m.group(2)) - 1)
                out[d.strftime("%Y%m%d")] = str(f)
        for f in sorted(self.tec_dir.glob("jplg*.*i")):
            m = re.search(r"jplg(\d{3})0\.(\d{2})i", f.name)
            if m:
                d = datetime(2000 + int(m.group(2)), 1, 1) + timedelta(days=int(m.group(1)) - 1)
                out[d.strftime("%Y%m%d")] = str(f)
        return out

    def _cslc_output_of(self, run_script: Path) -> Path:
        """The .h5 a ``run_<date>_<burst_id>.sh`` script is expected to write."""
        prefix = run_script.stem                # run_<date>_<burst_id>
        date_str = prefix.split("_")[1]
        burst_id = prefix.split(date_str, 1)[1][1:]
        return self.cslc_dir / burst_id / date_str / f"{burst_id}_{date_str}.h5"

    def run_cslc(self, force: bool = False, prepare_only: bool = False) -> bool:
        """Geocode every burst x date onto a common UTM grid.

        Three sub-steps, from the stack notebook's sections 2.1-2.3:

          1. ``s1_geocode_stack.py`` -> ``cslc/run_files/`` + ``cslc/runconfigs/``
          2. inject each date's IONEX file into its runconfig, so COMPASS applies
             the ionospheric timing correction (a runconfig with a null
             ``tec_file`` silently drops the term rather than failing)
          3. execute the run_files, skipping any whose output .h5 already exists

        ``--common-bursts-only`` keeps only bursts present on EVERY date. That
        guarantees a rectangular stack, but silently narrows coverage when one
        date is missing a burst -- so the burst count is reported here rather
        than left for the user to notice later.
        """
        if not self._ensure_burst_db():
            return False
        for p, what in ((self.slc_dir, "SLC"), (self.dem_path, "DEM"),
                        (self.orbit_dir, "orbits")):
            if not p.exists():
                logger.error("ISCE3_Burst: %s not found at %s", what, p)
                return False
        # Geocode extent: under process_full_extent this is the full burst
        # footprint (from the .SAFE files); otherwise config.AOI. Going through
        # _aoi() keeps the cslc --bbox, the DEM footprint and every later stage
        # measuring the SAME extent.
        try:
            aoi = self._aoi()
        except Exception as exc:                                 # noqa: BLE001
            logger.error("ISCE3_Burst: cannot determine a geocode extent for "
                         "--bbox (%s)", exc)
            return False
        if not aoi or len(aoi) != 4:
            logger.error("ISCE3_Burst: could not resolve a (W,S,E,N) extent for --bbox")
            return False

        env = self._env()
        cslc = self.cslc_dir
        cslc.mkdir(parents=True, exist_ok=True)
        run_dir, cfg_dir = cslc / "run_files", cslc / "runconfigs"

        # ── 1. generate run files + runconfigs ─────────────────────────────
        if run_dir.exists() and any(run_dir.glob("run_*.sh")) and not force:
            print(f"[ISCE3_Burst] run_files already present ({len(list(run_dir.glob('run_*.sh')))}) "
                  f"-- use force=True to regenerate")
        else:
            # Same argument order as the notebook, built explicitly rather than
            # by index arithmetic -- an insert(-5, ...) silently lands in the
            # wrong place the moment the list changes.
            cmd = ["s1_geocode_stack.py",
                   "-s", str(self.slc_dir), "-d", str(self.dem_path),
                   "-o", str(self.orbit_dir), "-w", str(cslc),
                   "-dx", str(getattr(self.config, "x_posting", 10)),
                   "-dy", str(getattr(self.config, "y_posting", 20))]
            if getattr(self.config, "common_bursts_only", True):
                cmd.append("--common-bursts-only")
            cmd += ["--burst-db-file", str(self.burst_db_path), "--unzipped",
                    "--bbox", *[str(x) for x in aoi]]
            print(f"  $ {' '.join(cmd)}")
            p = subprocess.run(cmd, env=env, text=True,
                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
            if p.returncode != 0:
                logger.error("ISCE3_Burst: s1_geocode_stack.py failed (rc=%s): %s",
                             p.returncode, (p.stdout or "")[-800:])
                return False

        runs = sorted(run_dir.glob("run_*.sh"))
        if not runs:
            logger.error("ISCE3_Burst: s1_geocode_stack.py produced no run_files in %s", run_dir)
            return False
        bursts = sorted({r.stem.split("_", 2)[2] for r in runs if len(r.stem.split("_")) > 2})
        print(f"[ISCE3_Burst] {len(runs)} burst x date job(s), {len(bursts)} burst(s): {bursts}")

        # ── 2. inject TEC ──────────────────────────────────────────────────
        tmap = self.tec_map()
        print(f"[ISCE3_Burst] IONEX map: {len(tmap)} date(s)")
        if not tmap:
            logger.warning("ISCE3_Burst: no IONEX files in %s -- CSLC will be "
                           "generated WITHOUT the ionospheric correction", self.tec_dir)
        else:
            import yaml
            n_set = n_cfg = 0
            for cfg_path in sorted(cfg_dir.glob("geo_runconfig_????????_*.yaml")):
                n_cfg += 1
                parts = cfg_path.stem.split("_")
                if len(parts) < 4 or parts[2] not in tmap:
                    continue
                cfg = yaml.safe_load(cfg_path.read_text())
                grp = cfg["runconfig"]["groups"]["dynamic_ancillary_file_group"]
                if grp.get("tec_file") is None:
                    grp["tec_file"] = tmap[parts[2]]
                    cfg_path.write_text(yaml.dump(cfg, default_flow_style=False,
                                                  sort_keys=False))
                    n_set += 1
            print(f"[ISCE3_Burst] runconfigs given a tec_file: {n_set}/{n_cfg}")

        if prepare_only:
            # HPC path: the run scripts ARE the SLURM units, so generation and
            # TEC injection must finish before submission, but nothing is
            # executed here.
            for r in runs:
                r.chmod(r.stat().st_mode | 0o111)
            print(f"[ISCE3_Burst] prepared {len(runs)} run script(s) for submission")
            return True

        # ── 3. execute ─────────────────────────────────────────────────────
        todo = []
        for r in runs:
            out_h5 = self._cslc_output_of(r)
            if out_h5.exists() and not force:
                continue
            r.chmod(r.stat().st_mode | 0o111)
            todo.append((r, out_h5))
        done = len(runs) - len(todo)
        print(f"[ISCE3_Burst] {done} already generated, {len(todo)} to run")

        failures = []
        def _one(item):
            r, out_h5 = item
            p = subprocess.run(["bash", str(r)], env=env, text=True,
                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
            ok = p.returncode == 0 and out_h5.exists()
            if not ok:
                failures.append((r.name, (p.stdout or "")[-400:]))
            return ok

        nw = max(1, int(getattr(self.config, "max_workers", 3)))
        if todo:
            with ThreadPoolExecutor(max_workers=nw) as ex:
                for i, ok in enumerate(ex.map(_one, todo), 1):
                    print(f"  [{i}/{len(todo)}] {'ok' if ok else 'FAILED'}")
        for name, tail in failures:
            logger.error("ISCE3_Burst: %s failed: %s", name, tail)

        made = len(list(cslc.glob("t*/*/*.h5")))
        print(f"[ISCE3_Burst] CSLC .h5 on disk: {made} (expected {len(runs)})")
        if made < len(runs):
            return False
        if failures:
            # COMPASS writes the .h5 and only then renders a PNG browse image.
            # That last step needs GDAL's HDF5 driver, so on an installation
            # without it every run "fails" with a complete, correct product.
            # Judge the stage on the products, not the exit code.
            logger.warning("ISCE3_Burst: %d run script(s) exited non-zero but all "
                           "%d CSLC products exist -- treating as success; see "
                           "the logged tails for what failed after writing",
                           len(failures), made)
        return True

    # ------------------------------------------------------------------
    # dolphin engine: ifg / stitch / unwrap
    # ------------------------------------------------------------------

    def _aoi(self) -> list[float]:
        """Processing extent as (W, S, E, N) degrees.

        ``config.AOI`` is normally already populated -- ISCE3_Burst_Config's
        __post_init__ seeds it from the folder's downloader ``intersectsWith``
        -- so this usually just returns it. The fallbacks cover a config built
        before the folder had a downloader section, or one loaded from an older
        saved file.

        With ``process_full_extent`` set, AOI is ignored and the extent is
        measured from the geocoded bursts instead.

        The old behaviour required AOI and raised otherwise, so a folder
        downloaded by AOI -- which already recorded that AOI in the same
        directory -- still refused to process until the box was filled in by
        hand, and processing a whole burst footprint was not expressible.
        """
        if bool(getattr(self.config, "process_full_extent", False)):
            # Prefer geocoded bursts once they exist (exact output extent);
            # before then, measure the burst footprint straight from the .SAFE
            # files so dem/cslc actually cover the full extent rather than
            # silently shrinking to the AOI.
            box = self._aoi_from_bursts() or self._full_extent_from_safes()
            if box:
                print(f"[ISCE3_Burst] full downloaded extent: "
                      f"{['%.4f' % v for v in box]}")
                return box
            # Neither geocoded bursts nor readable .SAFE footprints -- fall back
            # to the AOI so the stage can still run rather than hard-failing.
            print(f"[ISCE3_Burst] process_full_extent set but could not measure "
                  f"the burst footprint (no geocoded bursts, no readable .SAFE) "
                  f"-- falling back to AOI for this stage")

        aoi = getattr(self.config, "AOI", None)
        if aoi and len(aoi) == 4:
            return [float(x) for x in aoi]

        box = _aoi_bbox_from_folder(self.workdir)
        if box:
            print(f"[ISCE3_Burst] AOI from the downloader's intersectsWith: "
                  f"{['%.4f' % v for v in box]}")
            return box

        box = self._aoi_from_bursts()
        if not box:
            raise ValueError(
                "ISCE3_Burst: could not determine a processing extent. No "
                "config.AOI, no intersectsWith in this folder's "
                "insarhub_config.json, and no geocoded bursts on disk to "
                "measure. Set AOI explicitly.")
        print(f"[ISCE3_Burst] AOI = full downloaded extent: "
              f"{['%.4f' % v for v in box]}")
        return box

    def _aoi_from_bursts(self) -> list[float] | None:
        """(W, S, E, N) covering every geocoded burst, in degrees.

        The CSLCs are geocoded to absolute UTM, so their bounds are reprojected
        to EPSG:4326 and unioned -- this is the true full extent of what was
        downloaded, which is what 'process the whole thing' has to mean.
        """
        try:
            from osgeo import gdal, osr
        except Exception:                                        # noqa: BLE001
            return None

        # The geocoded CSLC .h5 (written by `cslc`, before any crop) are the full
        # burst footprints on a common UTM grid. Do NOT fall back to a bare
        # cslc_dir *.slc.tif glob: the only *.slc.tif under cslc/ are COMPASS's
        # radar-coordinate SCRATCH intermediates with no map CRS.
        h5s = sorted(p for p in self.cslc_dir.glob("t*_iw*/*/*.h5")
                     if "static_layers" not in p.name)
        if not h5s:
            return None

        wgs = osr.SpatialReference()
        wgs.ImportFromEPSG(4326)
        wgs.SetAxisMappingStrategy(osr.OAMS_TRADITIONAL_GIS_ORDER)

        w = s = e = n = None
        for h5 in h5s:
            # A partially-written or corrupt raster (e.g. a crop that died
            # mid-write, or a stray leftover) must NOT abort extent measurement
            # -- skip it and use the rest. The ``if ds is None`` guard below only
            # catches failures when gdal.UseExceptions() is OFF; other ISCE3
            # stages turn it ON process-globally, and then gdal.Open RAISES
            # "OGR Error: Corrupt data" on a bad file instead of returning None,
            # which previously escaped here and failed the whole crop stage.
            try:
                ds = gdal.Open(f'NETCDF:"{h5}":/data/VV')
                if ds is None:
                    continue
                gt, nx, ny = ds.GetGeoTransform(), ds.RasterXSize, ds.RasterYSize
                src = osr.SpatialReference()
                src.ImportFromWkt(ds.GetProjection())
                src.SetAxisMappingStrategy(osr.OAMS_TRADITIONAL_GIS_ORDER)
                tf = osr.CoordinateTransformation(src, wgs)
                xs = (gt[0], gt[0] + gt[1] * nx)
                ys = (gt[3], gt[3] + gt[5] * ny)
                for x in xs:
                    for y in ys:
                        lon, lat, *_ = tf.TransformPoint(x, y)
                        w = lon if w is None else min(w, lon)
                        e = lon if e is None else max(e, lon)
                        s = lat if s is None else min(s, lat)
                        n = lat if n is None else max(n, lat)
            except Exception as exc:                              # noqa: BLE001
                logger.warning("ISCE3_Burst: skipping unreadable CSLC %s while "
                               "measuring the full extent (%s)", h5.name, exc)
                continue
            finally:
                ds = None

        return None if w is None else [w, s, e, n]

    def _full_extent_from_safes(self) -> list[float] | None:
        """(W, S, E, N) covering every downloaded burst, from the .SAFE burst
        footprints -- the PRE-geocode source for ``process_full_extent``.

        ``_aoi_from_bursts`` can only measure geocoded output, which does not
        exist until after cslc, so before then it returns None and the extent
        would otherwise fall back to the (smaller) AOI -- silently defeating
        process_full_extent for the two stages that define the footprint (dem,
        cslc). A burst stack is the SAME burst IDs on every date, so one SAFE's
        burst borders already describe the whole stack's ground footprint: this
        reads the first readable SAFE, unions its bursts across all subswaths,
        and caches the result. Borders come from the annotation geolocation
        grid, so no orbit file is needed.
        """
        cached = getattr(self, "_full_extent_cache", None)
        if cached is not None:
            return cached or None            # a miss is cached as [] -> None
        box = None
        try:
            import warnings
            import s1reader
            from shapely.ops import unary_union
            for safe in sorted(self.slc_dir.glob("*.SAFE")):
                polys = []
                for sw in (1, 2, 3):
                    try:
                        with warnings.catch_warnings():
                            warnings.simplefilter("ignore")
                            polys += [b.border for b in
                                      s1reader.load_bursts(str(safe), None, sw)]
                    except Exception:                            # noqa: BLE001
                        continue
                if polys:
                    w, s, e, n = unary_union(polys).bounds
                    box = [float(w), float(s), float(e), float(n)]
                    break
        except Exception as exc:                                  # noqa: BLE001
            logger.warning("ISCE3_Burst: could not read burst footprints from "
                           "the .SAFE files for process_full_extent (%s)", exc)
        self._full_extent_cache = box or []
        return box

    @property
    def geom_dir(self) -> Path:
        """LOS/geometry rasters on the interferogram grid, for SBAS."""
        return self.workdir / "geometry"

    def run_static(self, force: bool = False) -> bool:
        """Static geometry layers per burst, then LOS vectors on the stack grid.

        The geometry does not change with time -- it comes from the orbit, the
        DEM and the burst's own timing -- so this runs once per burst, on that
        burst's earliest runconfig, not once per acquisition.

        Two steps: COMPASS's ``s1_static_layers.py`` writes
        ``static_layers_<burst>.h5`` (los_east, los_north, local_incidence_angle,
        layover_shadow_mask), then those are cropped, mosaicked and resampled
        onto the interferogram grid so SBAS can actually use them.
        """

        cfg_dir = self.cslc_dir / "runconfigs"
        cfgs = sorted(cfg_dir.glob("geo_runconfig_????????_*.yaml"))
        if not cfgs:
            print(f"[ISCE3_Burst] no runconfigs in {cfg_dir} -- run 'cslc' first")
            return False

        # earliest runconfig per burst
        per_burst: dict[str, Path] = {}
        for f in cfgs:
            parts = f.stem.split("_")
            if len(parts) < 4:
                continue
            date_str = parts[2]
            burst_id = f.stem.split(date_str, 1)[1][1:]
            per_burst.setdefault(burst_id, f)

        env = self._env()
        todo = []
        for burst_id, cfg in sorted(per_burst.items()):
            hits = list(self.cslc_dir.glob(
                f"{burst_id}/*/static_layers_{burst_id}.h5"))
            if hits and not force:
                continue
            todo.append((burst_id, cfg))
        print(f"[ISCE3_Burst] static layers: {len(per_burst) - len(todo)} present, "
              f"{len(todo)} to generate")

        failures = []
        for burst_id, cfg in todo:
            p = subprocess.run(["s1_static_layers.py", str(cfg)], env=env,
                               text=True, stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT)
            hits = list(self.cslc_dir.glob(
                f"{burst_id}/*/static_layers_{burst_id}.h5"))
            if not hits:
                failures.append((burst_id, (p.stdout or "")[-400:]))
                logger.error("ISCE3_Burst: static layers failed for %s: %s",
                             burst_id, (p.stdout or "")[-400:])
            else:
                print(f"  {burst_id}: {hits[0].name}")
        if failures:
            return False

        # put them on the grid the unwrapped products live on
        like = sorted(self.stitch_dir.glob("*.int.tif"))
        if not like:
            print("[ISCE3_Burst] static layers generated, but no stitched "
                  "interferogram to define the output grid -- run 'stitch' to "
                  "get los_*.tif on the stack grid")
            return True

        made = build_los_layers(
            self.cslc_dir, self.geom_dir, self._aoi(), like[0],
            buffer_deg=0.05)
        print(f"[ISCE3_Burst] geometry on stack grid: "
              f"{sorted(made)} -> {self.geom_dir}")
        return "los_up" in made

    def run_los(self, force: bool = False) -> bool:
        """Put the static geometry layers on the interferogram grid.

        Separate from `static` because it needs a stitched interferogram to
        resample onto -- see the STAGES comment. Produces los_east/los_north/
        los_up and incidence in geometry/, which the analyzer needs to convert
        line-of-sight displacement into a ground component.
        """
        return self._run_static_los()

    def _cslc_list(self) -> list[str]:
        """All geocoded CSLC ``.h5`` (excluding static layers), sorted."""
        cslc = sorted(self.cslc_dir.glob("t*_iw*/*/t*_iw*_2*.h5"))
        cslc = [str(p) for p in cslc if "static_layers" not in p.name]
        if not cslc:
            raise FileNotFoundError(
                f"ISCE3_Burst: no CSLC HDF5 under {self.cslc_dir}; "
                "run the 'cslc' stage first")
        return cslc

    #: HDF5 subdataset dolphin reads the complex SLC from. COMPASS CSLC stores
    #: it at ``/data/VV``; a NISAR-GSLC subclass overrides this with the NISAR
    #: GSLC grid path (``/science/LSAR/GSLC/grids/frequencyA/<pol>``).
    _subdataset: str = "/data/VV"

    def _dolphin_cfg(self):
        """Build dolphin's ``DisplacementWorkflow`` from this processor's config.

        The single source of truth for every dolphin-engine stage (ifg/stitch/
        unwrap). ``process_full_extent=True`` leaves ``bounds`` unset so dolphin
        uses the burst union (exactly ``dolphin run``); a specific AOI clips the
        stitched output via ``bounds``.
        """
        from dolphin.workflows.config import DisplacementWorkflow

        c = self.config
        out_opts: dict = {"strides": {"x": int(getattr(c, "rglks", 4)),
                                      "y": int(getattr(c, "azlks", 2))}}
        if not bool(getattr(c, "process_full_extent", False)):
            aoi = self._aoi()
            if aoi and len(aoi) == 4:
                out_opts["bounds"] = [float(x) for x in aoi]
                out_opts["bounds_epsg"] = 4326
        wmask = self.water_mask_path if self.water_mask_path.exists() else None
        # Full phase-linking option set, so every pl_* config field is honoured
        # (matches the native `dolphin config` defaults for the unset ones).
        phase_linking: dict = {
            "ministack_size": int(getattr(c, "pl_ministack_size", 15)),
            "half_window": {"x": int(getattr(c, "pl_half_window_x", 14)),
                            "y": int(getattr(c, "pl_half_window_y", 7))},
            "shp_method": str(getattr(c, "pl_shp_method", "glrt")),
            "shp_alpha": float(getattr(c, "pl_shp_alpha", 0.001)),
            "use_evd": bool(getattr(c, "pl_use_evd", False)),
            "beta": float(getattr(c, "pl_beta", 0.0)),
            "baseline_lag": getattr(c, "pl_baseline_lag", None),
        }
        # network after phase linking: 'single_reference' -> output_reference_idx,
        # 'bandwidth' (default, matches dolphin) -> max_bandwidth = n_connections;
        # max_temporal_baseline (days) wins over n_connections when set.
        net_opts: dict = {}
        if str(getattr(c, "pl_ifg_network", "bandwidth")).lower() == "single_reference":
            net_opts["reference_idx"] = 0
        elif getattr(c, "max_temporal_baseline", None):
            net_opts["max_temporal_baseline"] = float(c.max_temporal_baseline)
        else:
            net_opts["max_bandwidth"] = int(getattr(c, "n_connections", 3))
        t = int(getattr(c, "unwrap_tiles", 1))
        unwrap_opts: dict = {
            "run_unwrap": True,
            "unwrap_method": str(getattr(c, "unw_method", "snaphu")),
            "n_parallel_jobs": max(1, int(self._nw)),
            "snaphu_options": {
                "ntiles": (t, t),
                "tile_overlap": (0, 0),
                "init_method": str(getattr(c, "unwrap_init_method", "mcf")),
                "cost": str(getattr(c, "unwrap_cost", "smooth")),
            },
        }
        return DisplacementWorkflow(
            cslc_file_list=self._cslc_list(),
            input_options={"subdataset": self._subdataset},
            work_directory=str(self.workdir),
            mask_file=str(wmask) if wmask else None,
            output_options=out_opts,
            phase_linking=phase_linking,
            interferogram_network=net_opts,
            unwrap_options=unwrap_opts,
            worker_settings={
                "threads_per_worker": max(1, int(getattr(c, "num_threads", 4))),
                "n_parallel_bursts": 1,
            },
        )

    def _wrapped_phase_manifest_path(self) -> Path:
        return self.workdir / "interferograms" / "wrapped_phase_manifest.json"

    def run_ifg(self, force: bool = False) -> bool:
        """PS + phase-link + interferograms per burst (dolphin ``wrapped_phase.run``).

        One ``wrapped_phase.run`` call per burst, exactly as dolphin's
        ``displacement.run`` does. Outputs land in dolphin's native per-burst
        layout (``<workdir>/<burst_id>/linked_phase/``, ``interferograms/``,
        ``PS/``); the per-burst file lists are recorded to a manifest that the
        ``stitch`` stage consumes.
        """
        ensure_proj_env()
        ensure_gdal_cli()          # dolphin's stitch shells out to gdal_merge.py
        from dolphin.workflows import wrapped_phase
        from dolphin.workflows._utils import _create_burst_cfg, _remove_dir_if_empty
        from opera_utils import group_by_burst

        cfg = self._dolphin_cfg()
        grouped = group_by_burst(cfg.cslc_file_list)
        if not grouped:
            raise FileNotFoundError(
                f"ISCE3_Burst: no bursts from {len(cfg.cslc_file_list)} CSLC(s)")

        empty = {b: [] for b in grouped}
        ifg_files: list[str] = []
        temp_coh: list[str] = []
        ps: list[str] = []
        crlb: list[str] = []
        closure: list[str] = []
        amp_disp: list[str] = []
        shp: list[str] = []
        sim: list[str] = []
        for burst in sorted(grouped):
            burst_cfg = _create_burst_cfg(cfg, burst, grouped, empty, empty, empty)
            # Create the per-burst work dir (and subdirs) first -- dolphin's
            # displacement.run does this before wrapped_phase.run, and it also
            # removes the empty timeseries/unwrapped dirs re-grouping leaves.
            burst_cfg.create_dir_tree()
            _remove_dir_if_empty(burst_cfg.timeseries_options._directory)
            _remove_dir_if_empty(burst_cfg.unwrap_options._directory)
            print(f"[ISCE3_Burst] ifg: {burst} ({len(grouped[burst])} CSLC) "
                  f"-> wrapped_phase.run")
            out = wrapped_phase.run(burst_cfg, max_workers=self._nw)
            ifg_files += [str(p) for p in out.ifg_file_list]
            temp_coh += [str(p) for p in out.temp_coh_files]
            ps.append(str(out.ps_looked_file))
            crlb += [str(p) for p in out.crlb_files]
            closure += [str(p) for p in out.closure_phase_files]
            amp_disp.append(str(out.amp_disp_looked_file))
            shp += [str(p) for p in out.shp_count_files]
            sim += [str(p) for p in out.similarity_files]

        import json as _json
        manifest = self._wrapped_phase_manifest_path()
        manifest.parent.mkdir(parents=True, exist_ok=True)
        manifest.write_text(_json.dumps({
            "ifg_file_list": ifg_files,
            "temp_coh_file_list": temp_coh,
            "ps_file_list": ps,
            "crlb_file_list": crlb,
            "closure_phase_file_list": closure,
            "amp_dispersion_list": amp_disp,
            "shp_count_file_list": shp,
            "similarity_file_list": sim,
        }, indent=1))
        print(f"[ISCE3_Burst] ifg: {len(ifg_files)} per-burst interferogram(s) "
              f"over {len(grouped)} burst(s)")
        return bool(ifg_files)

    # ------------------------------------------------------------------
    # HPC decomposition
    # ------------------------------------------------------------------

    def stage_progress(self, stage: str) -> tuple[int, int]:
        """``(done, total)`` for a stage, from PRODUCTS ON DISK.

        Deliberately independent of how the stage was run. Counting a
        submission's SLURM jobs describes only the last invocation: a resume
        that repaired 19 of 192 burst-dates reported "19/19", which is true of
        that submission and useless as a description of the stack. It also made
        a finished stage look unfinished whenever leftover child scripts from a
        longer earlier list inflated the denominator.

        Products are the only measure that is the same whether a stage ran
        locally, on SLURM, in one go or across five resumes -- so this is what
        "is the workdir done?" actually means.

        ``(0, 0)`` means "no meaningful count", not "nothing done".
        """
        try:
            if stage == "dem":
                return (1 if self.dem_path.exists() else 0, 1)
            if stage == "tec":
                dates = set(self.acquisition_dates())
                have = _ionex_dates_on_disk(self.tec_dir) & dates
                return (len(have), len(dates))
            if stage == "cslc":
                runs = sorted((self.cslc_dir / "run_files").glob("run_*.sh"))
                done = sum(1 for r in runs if self._cslc_output_of(r).exists())
                return (done, len(runs))
            if stage == "static":
                n = len(self._static_runconfigs())
                have = len([f for f in self.cslc_dir.rglob("static_layers_*.h5")])
                return (have, n)
            if stage == "los":
                want = ["los_east", "los_north", "los_up",
                        "local_incidence_angle", "layover_shadow_mask"]
                have = sum(1 for w in want if (self.geom_dir / f"{w}.tif").exists())
                return (have, len(want))
            if stage == "ifg":
                bursts = self.burst_dirs()
                done = sum(1 for b in bursts if (b / "interferograms").is_dir())
                return (done, len(bursts))
            if stage == "stitch":
                n = len(self._pairs())
                return (len(list(self.stitch_dir.glob("*.int.tif"))), n)
            if stage == "unwrap":
                n = len(self._pairs())
                return (len(list(self.unwrap_dir.glob("*.unw.tif"))), n)
        except Exception as exc:                                 # noqa: BLE001
            logger.debug("stage_progress(%s) failed: %s", stage, exc)
        return (0, 0)

    def _pairs(self) -> list[str]:
        """Unique ``YYYYMMDD_YYYYMMDD`` pairs, from the ifg stage's manifest."""
        import json as _json
        m = self._wrapped_phase_manifest_path()
        if not m.exists():
            return []
        try:
            data = _json.loads(m.read_text())
        except Exception:                                        # noqa: BLE001
            return []
        pairs = set()
        for f in data.get("ifg_file_list", []):
            mm = re.search(r"(\d{8})_(\d{8})", Path(f).name)
            if mm:
                pairs.add(f"{mm.group(1)}_{mm.group(2)}")
        return sorted(pairs)

    def stage_units(self, stage: str) -> list[tuple[str, bool]]:
        """``[(unit label, done), ...]`` for every real unit of a stage.

        What ``--ls`` lists. Like :meth:`stage_progress` this is derived from
        products, not from the last submission's ``cmd_XXXX`` markers -- those
        cover only whatever that invocation happened to run, so after a resume
        that repaired 19 dates ``--ls cslc`` listed 19 lines for a 192-date
        stage. Labels are the real identifiers (date + burst, or the pair) so a
        missing unit can be acted on directly.
        """
        try:
            if stage == "cslc":
                return [(r.stem.removeprefix("run_"),
                         self._cslc_output_of(r).exists())
                        for r in sorted((self.cslc_dir / "run_files").glob("run_*.sh"))]
            if stage == "tec":
                have = _ionex_dates_on_disk(self.tec_dir)
                return [(d, d in have) for d in sorted(set(self.acquisition_dates()))]
            if stage == "static":
                return [(c.stem.split("_", 3)[-1],
                         bool(list(self.cslc_dir.rglob(
                             f"static_layers_{c.stem.split('_', 3)[-1]}.h5"))))
                        for c in self._static_runconfigs()]
            if stage == "dem":
                return [(self.dem_path.name, self.dem_path.exists())]
            if stage == "los":
                return [(w, (self.geom_dir / f"{w}.tif").exists())
                        for w in ("los_east", "los_north", "los_up",
                                  "local_incidence_angle", "layover_shadow_mask")]
            if stage == "ifg":
                return [(b.name, (b / "interferograms").is_dir())
                        for b in self.burst_dirs()]
            if stage in ("stitch", "unwrap"):
                d, suf = {"stitch": (self.stitch_dir, ".int.tif"),
                          "unwrap": (self.unwrap_dir, ".unw.tif")}[stage]
                return [(p, (d / f"{p}{suf}").exists()) for p in self._pairs()]
        except Exception as exc:                                 # noqa: BLE001
            logger.debug("stage_units(%s) failed: %s", stage, exc)
        return []

    def _reentry(self, stage: str, index: int | None = None) -> str:
        """Shell command running ONE unit of a stage by re-entering insarhub.

        Most stages are Python methods calling dolphin in-process, so a child
        job cannot be a bare shell line the way ISCE2's run-file commands are.
        Re-entering the CLI keeps the orchestration in one place instead of
        re-implementing each stage in bash -- the alternative that produced
        drift bugs elsewhere in this codebase.
        """
        exe = f"{sys.executable} -m insarhub.cli.main"
        c = (f'{exe} processor -N {type(self).name} -w "{self.workdir}" '
             f'run-stage-unit --stage {stage}')
        return c if index is None else f"{c} --index {index}"

    def hpc_phases(self, stage: str) -> list[tuple[str, list[str]]]:
        """How each stage splits into parallel SLURM child jobs.

        ==========  ==========================================================
        dem/tec     one job each -- network-bound, minutes, nothing to split
        cslc        one job per ``run_*.sh`` COMPASS already wrote. These are
                    real shell scripts, so no re-entry is needed and this is
                    exactly ISCE2's shape. The long pole: N dates x M bursts.
        static      map over bursts, then a 1-job reduce that mosaics the LOS
                    layers onto the interferogram grid
        ifg         one job -- dolphin's wrapped_phase.run loops all bursts
        stitch      one job -- dolphin's stitching_bursts.run is monolithic
        unwrap      one job -- dolphin's unwrapping.run is monolithic
        los         one job
        ==========  ==========================================================

        ``cslc`` needs its runconfigs to exist before the units can be listed, so
        the generation half runs here, synchronously, before submission -- the
        analogue of ISCE2 generating run files up front.
        """
        if stage in ("dem", "tec"):
            return [("", [self._reentry(stage)])]

        if stage == "cslc":
            runs = self._prepare_cslc_runs()
            return [("", [f'bash "{r}"' for r in runs])]

        if stage == "static":
            n = len(self._static_runconfigs())
            return [("", [self._reentry("static", i) for i in range(n)])]

        if stage == "los":
            return [("", [self._reentry("los")])]

        if stage in ("ifg", "stitch", "unwrap"):
            # dolphin's wrapped_phase.run / stitching_bursts.run / unwrapping.run
            # are each a single monolithic call -- no safe per-burst/pair split.
            return [("", [self._reentry(stage)])]

        return []

    def _static_runconfigs(self) -> list[Path]:
        """One runconfig per burst -- its earliest date, geometry being static."""
        cfgs = sorted((self.cslc_dir / "runconfigs").glob("geo_runconfig_????????_*.yaml"))
        first: dict[str, Path] = {}
        for c in cfgs:
            burst = c.stem.split("_", 3)[-1]
            first.setdefault(burst, c)
        return [first[k] for k in sorted(first)]

    def _prepare_cslc_runs(self) -> list[Path]:
        """Generate runconfigs + run_*.sh and return the scripts still to run.

        Runs synchronously before any job is submitted -- the analogue of
        ISCE2 generating its run files up front. The units of the cslc stage
        ARE these scripts, so they must exist before the manager can be built.
        Already-generated products are filtered out so a resumed submission
        only queues the missing dates.
        """
        if not self.run_cslc(prepare_only=True):
            raise RuntimeError(
                "ISCE3_Burst: could not generate COMPASS run files -- see the "
                "log above. Fix that before submitting to SLURM.")
        runs = sorted((self.cslc_dir / "run_files").glob("run_*.sh"))
        todo = [r for r in runs if not self._cslc_output_of(r).exists()]
        print(f"[ISCE3_Burst] cslc: {len(runs) - len(todo)} already geocoded, "
              f"{len(todo)} to submit")
        return todo

    def _run_static_unit(self, cfg: Path) -> bool:
        """One burst's static layers -- ``s1_static_layers.py <runconfig>``."""
        p = subprocess.run(["s1_static_layers.py", str(cfg)], env=self._env(),
                           text=True, stdout=subprocess.PIPE,
                           stderr=subprocess.STDOUT)
        if p.returncode != 0:
            logger.error("ISCE3_Burst: s1_static_layers.py %s failed: %s",
                         cfg.name, (p.stdout or "")[-500:])
            return False
        return True

    def _run_static_los(self) -> bool:
        """The reduce half of `static`: mosaic LOS layers onto the ifg grid.

        Deferring is NOT a failure. `static` sits before `stitch` in the stage
        order, but the grid it must resample onto is defined by the stitched
        interferograms -- so on a first pass through the chain there is nothing
        to resample onto yet. run_static() has always handled that by returning
        success with a note; this mirrors it. Re-run `static` once `stitch` is
        done to get los_*.tif.
        """
        like = sorted(self.stitch_dir.glob("*.int.tif"))
        if not like:
            print("[ISCE3_Burst] static layers generated, but no stitched "
                  "interferogram to define the output grid -- re-run 'static' "
                  "after 'stitch' to get los_*.tif on the stack grid")
            return True
        made = build_los_layers(
            self.cslc_dir, self.geom_dir, self._aoi(), like[0], buffer_deg=0.05)
        print(f"[ISCE3_Burst] geometry on stack grid: {sorted(made)}")
        return "los_up" in made

    def run_stage_unit(self, stage: str, index: int | None = None) -> bool:
        """Execute ONE unit of a stage. The entry point every child job calls.

        Deliberately no status-marker writes: a unit is a fraction of a stage,
        and the stage's verdict belongs to its manager, which writes SUCCEEDED
        or FAILED once every unit has finished. A unit reports via its exit
        code, which the child sbatch wrapper turns into a .done/.fail marker.
        """
        if stage == "static" and index is not None:
            return self._run_static_unit(self._static_runconfigs()[int(index)])
        if stage in ("los", "static_los"):   # static_los: pre-split alias
            return self._run_static_los()
        runner = getattr(self, f"run_{stage}", None)
        if runner is None:
            raise ValueError(f"ISCE3_Burst: no unit runner for stage {stage!r}")
        return bool(runner())

    def run_stitch(self, force: bool = False) -> bool:
        """Mosaic per-burst ifgs + estimate correlation (dolphin ``stitching_bursts.run``)."""
        ensure_proj_env()
        ensure_gdal_cli()          # dolphin's stitch shells out to gdal_merge.py
        import json as _json
        from dolphin.workflows import stitching_bursts

        manifest = self._wrapped_phase_manifest_path()
        if not manifest.exists():
            raise FileNotFoundError(
                f"ISCE3_Burst: {manifest} missing -- run the 'ifg' stage first")
        m = _json.loads(manifest.read_text())
        cfg = self._dolphin_cfg()
        stitched = stitching_bursts.run(
            ifg_file_list=[Path(p) for p in m["ifg_file_list"]],
            temp_coh_file_list=[Path(p) for p in m["temp_coh_file_list"]],
            ps_file_list=[Path(p) for p in m["ps_file_list"]],
            crlb_file_list=[Path(p) for p in m["crlb_file_list"]],
            closure_phase_file_list=[Path(p) for p in m["closure_phase_file_list"]],
            amp_dispersion_list=[Path(p) for p in m["amp_dispersion_list"]],
            shp_count_file_list=[Path(p) for p in m["shp_count_file_list"]],
            similarity_file_list=[Path(p) for p in m["similarity_file_list"]],
            stitched_ifg_dir=cfg.interferogram_network._directory,
            output_options=cfg.output_options,
            file_date_fmt=cfg.input_options.cslc_date_fmt,
            corr_window_size=(11, 11),
            num_workers=self._nw,
        )
        print(f"[ISCE3_Burst] stitch: {len(stitched.ifg_paths)} stitched ifgs, "
              f"{len(stitched.interferometric_corr_paths)} correlations -> "
              f"{cfg.interferogram_network._directory}")
        return (bool(stitched.ifg_paths)
                and len(stitched.interferometric_corr_paths) >= len(stitched.ifg_paths))

    def run_unwrap(self, force: bool = False) -> bool:
        """Unwrap stitched ifgs with snaphu (dolphin ``unwrapping.run``)."""
        ensure_proj_env()
        from dolphin.workflows import unwrapping

        cfg = self._dolphin_cfg()
        stitched_dir = cfg.interferogram_network._directory
        ifg_files = sorted(stitched_dir.glob("*.int.tif"))
        cor_files = sorted(stitched_dir.glob("*.int.cor.tif"))
        if not ifg_files:
            raise FileNotFoundError(
                f"ISCE3_Burst: no stitched .int.tif under {stitched_dir}; "
                "run the 'stitch' stage first")
        mask = self.water_mask_path if self.water_mask_path.exists() else None
        if mask is None:
            logger.warning("ISCE3_Burst: no water mask at %s -- unwrapping over "
                           "water, which can seed errors that leak inland",
                           self.water_mask_path)
        unwrapped, conncomp = unwrapping.run(
            ifg_file_list=ifg_files,
            cor_file_list=cor_files,
            nlooks=self._unwrap_nlooks(),
            unwrap_options=cfg.unwrap_options,
            temporal_coherence_filename=self.quality_file(),
            similarity_filename=None,
            mask_file=str(mask) if mask else None,
        )
        print(f"[ISCE3_Burst] unwrap: {len(unwrapped)} unwrapped, "
              f"{len(conncomp)} conncomp -> {cfg.unwrap_options._directory}")
        return len(unwrapped) >= len(ifg_files)

    def _unwrap_nlooks(self) -> float:
        """Effective looks handed to snaphu, derived exactly as dolphin does.

        dolphin's displacement workflow computes ``nlooks = (2*hw_y+1)*(2*hw_x+1)``
        (15*29 = 435 at the defaults) -- the SHP window *is* the adaptive
        multilook. A configured ``unwrap_nlooks`` overrides it.
        """
        v = getattr(self.config, "unwrap_nlooks", None)
        if v:
            return float(v)
        return float(
            (2 * int(getattr(self.config, "pl_half_window_y", 7)) + 1)
            * (2 * int(getattr(self.config, "pl_half_window_x", 14)) + 1))

从 NISAR L2 GSLC 数据构建干涉图栈，使用 dolphin 完成相位链接、干涉图与解缠。请搭配 NISAR_GSLC 下载器与 ISCE3_Dolphin_NISAR_PL 分析器。

与 ISCE3_Burst 使用相同的 dolphin 引擎，但不做地理编码——GSLC 本身已完成地理编码且每个日期一帧，因此 dem/tec/cslc/static 阶段被去掉，ifg 是对整个栈的单次 wrapped_phase.run。

三个阶段，依次运行：

阶段	工具	输出
`ifg`	dolphin	PS + 相位链接 + 干涉图（对整个栈一次 `wrapped_phase.run`）
`stitch`	dolphin	相干性 + 镶嵌（每日期单帧时几乎为空操作）
`unwrap`	snaphu	解缠相位 + 连通分量

为什么这里需要 AOI 裁剪

一帧 NISAR GSLC 极大（如 69840 × 68688 像素），而 AOI 通常只是很小的窗口。ISCE3_Burst 通过 COMPASS 地理编码的 --bbox 免费获得 AOI 裁剪；ISCE3_NISAR 没有地理编码步骤，因此自行裁剪：在 ifg 阶段把每个 GSLC 的复数 SLC 子数据集按 AOI 裁成一个轻量 VRT（gdal_translate -of VRT -projwin，缓存于 workdir/cropped_gslc/），dolphin 只对该窗口做相位链接。

若不裁剪，dolphin 会处理整帧，stitch 的 gdal_merge 会在整帧栅格上耗尽内存（AOI 否则只会作为最后的输出裁剪，为时已晚）。在真实数据上，裁剪把每个输入从 69840 × 68688 降到约 8953 × 8728（缩小约 60×），把约 12 GB 的内存峰值降到几百 MB。勾选 process_full_extent 可禁用裁剪、处理整帧（需要大内存宿主机）。

配置

nisar_frequency（默认 A / B）与 nisar_polarization（默认 HH，HV、VV、VH）选择 dolphin 读取的 GSLC 网格组 —— /science/LSAR/GSLC/grids/frequency<freq>/<pol>。在一个栈内应保持不变。
AOI 与 ISCE3_Burst 一样从下载器的 intersectsWith 自动填充；裁剪直接使用它。
pl_*、n_connections/max_temporal_baseline 与 unwrap_* 字段对相位链接、干涉图网络与 snaphu 的调节方式与 ISCE3_Burst 完全相同。

其他说明

GSLC 下载得到的是整幅已地理编码的帧 —— 裁剪发生在处理时而非下载时，因此 slc/ 中的 .h5 产品不会被改动，可在不同 AOI 间复用。
运行于 insarhub-isce3-dolphin 镜像（与 ISCE3_Burst / Dolphin_SBAS 相同）；与其他处理器一样，设置 container 即可在无本地 ISCE3/dolphin 安装的情况下运行。

Source code in src/insarhub/processor/isce3_nisar.py

class ISCE3_NISAR(ISCE3_Burst):
    name = "ISCE3_NISAR"
    description = (
        "NISAR L2 GSLC -> dolphin phase-linking + interferograms (no geocoding; "
        "GSLC is already geocoded). Feeds the ISCE3_Dolphin_NISAR_PL analyzer."
    )
    compatible_downloader = "NISAR_GSLC"
    compatible_analyzer = "ISCE3_Dolphin_NISAR_PL"
    default_config = ISCE3_NISAR_Config
    JOBS_FILE = "isce3_nisar_jobs.json"

    # GSLC is pre-geocoded: only the dolphin engine stages run, preceded by an
    # AOI crop (NISAR's stand-in for the COMPASS geocode --bbox that ISCE3_Burst
    # gets its AOI cut from -- see run_crop).
    STAGES = ("crop", "ifg", "stitch", "unwrap")
    _IMPLEMENTED = ("crop", "ifg", "stitch", "unwrap")
    REQUIRED_MODULES = ("dolphin",)

    #: NISAR_GSLC downloads .h5 granules, not S1_Burst's .SAFE dirs.
    input_glob = "*GSLC*.h5"

    # ------------------------------------------------------------------ paths
    @property
    def gslc_dir(self) -> Path:
        """Where the NISAR_GSLC downloader put the ``*.h5`` products
        (defaults to workdir/slc, matching ISCE3_Burst's slc_dir convention)."""
        return self._p("gslc_dir", "slc")

    def _gslc_grid_path(self) -> str:
        """GDAL subdataset path of the complex SLC inside a NISAR GSLC .h5."""
        freq = str(getattr(self.config, "nisar_frequency", "A")).upper()
        pol = str(getattr(self.config, "nisar_polarization", "HH")).upper()
        return f"/science/LSAR/GSLC/grids/frequency{freq}/{pol}"

    @property
    def _subdataset(self) -> str:  # type: ignore[override]
        """dolphin reads the complex SLC from the NISAR GSLC grid group -- UNLESS
        the inputs have been AOI-cropped to standalone VRTs (see _cslc_list),
        in which case each VRT is already a single-band raster and no subdataset
        selection applies. ``_cslc_list`` runs first in ``_dolphin_cfg`` and sets
        ``_cropped_to_aoi``, so this reflects the choice it made."""
        if getattr(self, "_cropped_to_aoi", False):
            return ""
        return self._gslc_grid_path()

    @property
    def _cropped_gslc_dir(self) -> Path:
        return self._paths.cropped_gslc_dir

    def _raw_gslc_list(self) -> list[Path]:
        """The downloaded NISAR GSLC products, sorted -- one geocoded SLC per
        date. Excludes the small QA/side products ASF ships alongside."""
        h5 = sorted(
            p for p in self.gslc_dir.glob("*GSLC*.h5")
            if "QA" not in p.name and "STATS" not in p.name)
        if not h5:
            raise FileNotFoundError(
                f"ISCE3_NISAR: no NISAR GSLC .h5 under {self.gslc_dir}; "
                "run the NISAR_GSLC downloader first")
        return h5

    def _aoi_crop_enabled(self) -> bool:
        """Crop each GSLC to the AOI before phase-linking, unless the user asked
        for the full extent. GSLC is delivered as a full geocoded frame (e.g.
        69840x68688), but the AOI is usually a small window -- without cropping,
        dolphin phase-links the ENTIRE frame and the stitch's gdal_merge OOMs on
        the full-frame rasters (the AOI is otherwise only applied as a final
        output clip, too late to save the intermediate memory)."""
        if bool(getattr(self.config, "process_full_extent", False)):
            return False
        try:
            aoi = self._aoi()
        except Exception:                                        # noqa: BLE001
            return False
        return bool(aoi and len(aoi) == 4)

    def _crop_gslc_to_aoi(self, gslc: Path, aoi) -> Path:
        """Write a VRT that windows the GSLC's complex-SLC subdataset to the AOI.

        A VRT (not a copy) keeps this cheap -- it just records the source
        subdataset + the pixel window; dolphin reads it as an already-cropped
        single-band raster. ``-projwin_srs EPSG:4326`` lets us pass the AOI in
        lon/lat while the GSLC is in its native UTM. Idempotent + atomic.
        """
        w, s, e, n = (float(x) for x in aoi)
        self._cropped_gslc_dir.mkdir(parents=True, exist_ok=True)
        vrt = self._cropped_gslc_dir / f"{gslc.stem}.vrt"
        if vrt.exists():
            return vrt
        src = f'NETCDF:"{gslc}":{self._gslc_grid_path()}'
        tmp = vrt.with_name(vrt.name + ".tmp")
        # -projwin is ulx uly lrx lry == W N E S.
        cmd = ["gdal_translate", "-q", "-of", "VRT",
               "-projwin_srs", "EPSG:4326",
               "-projwin", str(w), str(n), str(e), str(s), src, str(tmp)]
        r = subprocess.run(cmd, capture_output=True, text=True, env=os.environ.copy())
        if r.returncode != 0 or not tmp.exists():
            raise RuntimeError(
                f"ISCE3_NISAR: AOI crop failed for {gslc.name}: {r.stderr.strip()}")
        os.replace(tmp, vrt)
        return vrt

    def _cslc_list(self) -> list[str]:  # type: ignore[override]
        """The stack dolphin reads: AOI-cropped VRTs when cropping applies (built
        by the ``crop`` stage; built on demand here too so ``ifg`` still works if
        run standalone -- ``_crop_gslc_to_aoi`` is idempotent), else the raw
        GSLC .h5 frames."""
        raw = self._raw_gslc_list()
        if self._aoi_crop_enabled():
            aoi = self._aoi()
            vrts = [self._crop_gslc_to_aoi(p, aoi) for p in raw]
            self._cropped_to_aoi = True
            return [str(v) for v in vrts]
        self._cropped_to_aoi = False
        return [str(p) for p in raw]

    # ------------------------------------------------------------------ crop
    def run_crop(self, force: bool = False) -> bool:
        """AOI-crop each GSLC to a lightweight VRT before phase-linking.

        NISAR's stand-in for the COMPASS geocode ``--bbox`` that gives
        ``ISCE3_Burst`` its AOI cut for free: a NISAR GSLC arrives as a full
        geocoded frame (e.g. 69840x68688), so this windows each one's complex-SLC
        subdataset to the AOI (see ``_crop_gslc_to_aoi``) and dolphin then
        processes only that window. A trivial pass-through when
        ``process_full_extent`` is set or no AOI is resolvable -- the full frame
        is used and the stage still succeeds.
        """
        ensure_gdal_cli()  # gdal_translate must be on PATH
        raw = self._raw_gslc_list()
        if not self._aoi_crop_enabled():
            print(f"[ISCE3_NISAR] crop: process_full_extent / no AOI -- using the "
                  f"full frames ({len(raw)} GSLC), no crop")
            return True
        aoi = self._aoi()
        if force:
            shutil.rmtree(self._cropped_gslc_dir, ignore_errors=True)
        vrts = [self._crop_gslc_to_aoi(p, aoi) for p in raw]
        print(f"[ISCE3_NISAR] crop: {len(vrts)} GSLC -> AOI "
              f"{['%.4f' % float(x) for x in aoi]} in {self._cropped_gslc_dir} "
              f"(dolphin processes only the AOI window, not the full frame)")
        return len(vrts) == len(raw)

    # ------------------------------------------------------------------ ifg
    def run_ifg(self, force: bool = False) -> bool:  # type: ignore[override]
        """PS + phase-link + interferograms over the WHOLE GSLC stack in one
        ``wrapped_phase.run`` (NISAR is one frame per date -- no burst split).

        Writes the same manifest ISCE3_Burst's ``stitch`` consumes, so the
        inherited ``run_stitch``/``run_unwrap`` run unchanged.
        """
        ensure_proj_env()
        ensure_gdal_cli()          # dolphin's stitch shells out to gdal_merge.py
        from dolphin.workflows import wrapped_phase

        cfg = self._dolphin_cfg()
        cfg.create_dir_tree()
        print(f"[ISCE3_NISAR] ifg: {len(cfg.cslc_file_list)} GSLC "
              f"({self._subdataset}) -> wrapped_phase.run")
        out = wrapped_phase.run(cfg, max_workers=self._nw)

        manifest = self._wrapped_phase_manifest_path()
        manifest.parent.mkdir(parents=True, exist_ok=True)
        manifest.write_text(json.dumps({
            "ifg_file_list": [str(p) for p in out.ifg_file_list],
            "temp_coh_file_list": [str(p) for p in out.temp_coh_files],
            "ps_file_list": [str(out.ps_looked_file)],
            "crlb_file_list": [str(p) for p in out.crlb_files],
            "closure_phase_file_list": [str(p) for p in out.closure_phase_files],
            "amp_dispersion_list": [str(out.amp_disp_looked_file)],
            "shp_count_file_list": [str(p) for p in out.shp_count_files],
            "similarity_file_list": [str(p) for p in out.similarity_files],
        }, indent=1))
        print(f"[ISCE3_NISAR] ifg: {len(out.ifg_file_list)} interferogram(s)")
        return bool(out.ifg_file_list)

处理器

可用处理器

使用方法

选择估计方式 — ifg_mode

处理范围

其他说明

为什么这里需要 AOI 裁剪

配置

其他说明

选择估计方式 — `ifg_mode`