Processor

The InSARHub Processor module provides functionality specifically for interferogram processing.

• Import processor

  Import the Processor class to access all processor functionality

  from insarhub import Processor
  

• View available processors

  List all registered processors

  Processor.available()
  

Available Processors

Hyp3_S1ISCE2_S1GMTSAR_S1ISCE3_BurstISCE3_NISAR

The HyP3 InSAR processor is a cloud-based processing service provided by the ASF HyP3 system for generating interferograms from Sentinel-1 SAR data. InSARHub wrapped hyp3_sdk as one of its process backends.

The Hyp3_S1 specifically wraps insar_job in hyp3_sdk to provide InSAR SLC processing workflows.

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

Usage

• Create Processor with Parameters

  Initialize a processor instance with search criteria

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

  OR

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

  OR

  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
  

• Submit Jobs

  Submit InSAR jobs to HyP3 based on the current configuration.

  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.

• Refresh Jobs

  Refresh the status of all jobs.

  jobs = processor.refresh()
  

  Raises:

  Type	Description
  ValueError	If no jobs are loaded in memory.

• Retry Failed Jobs

  Retry all failed jobs by re-submitting them.

  jobs = processor.retry()
  

• Download Succeeded Jobs

  Download all succeeded jobs for all users.

  processor.download()
  

• Save Current Jobs

  Save the current job batch information to a JSON file.

  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.

• Watch Jobs

  Continuously monitor jobs and download completed outputs.

  processor.watch()
  

  Parameters:

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

• Load Saved Job

  Load a previously saved JSON file and resume work.

  processor = Processor.create('Hyp3_S1', saved_job_path='path/to/your/json/file.json')
  

  When loaded, you can resume checking or downloading jobs submitted to the HyP3 server.

The ISCE2_S1 processor runs ISCE2 stackSentinel locally to generate Sentinel-1 interferograms from downloaded SLC .SAFE files. It generates a numbered sequence of run scripts and executes them sequentially, parallelising independent commands within each step.

• Import processor

  from insarhub import Processor
  

• Create 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],   # [S, N, W, E]
  )
  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.

• Submit (local mode)

  Generate run scripts and start sequential execution in a background process. Returns immediately; use refresh() to monitor progress.

  jobs = processor.submit()
  

• Submit (HPC / SLURM mode)

  Set hpc_mode=True to use the sliding-window SLURM manager. Steps are first grouped: consecutive steps with equal per-scene/per-pair command counts are merged into a single group-manager (e.g. run_02_unpack_secondary_slc + run_03_average_baseline when both have one command per scene); every other step gets its own single-step manager. Each manager keeps at most max_concurrent_hpc child jobs active at all times, submitting new ones immediately as slots open. Each sbatch script logs START/DONE/FAIL with elapsed seconds per command.

  Only the first group's manager is submitted directly by submit(). Every manager chain-submits the next group's manager itself right after it succeeds — via its own trailing sbatch call, not a SLURM --dependency — so at most one manager (plus its own ≤max_concurrent_hpc children) is ever sitting in the queue at a time, instead of every group's manager being pre-submitted up front. This matters because SLURM's submitted-jobs-per-user QOS limit counts jobs that are merely waiting on a dependency just as much as running ones; pre-submitting the whole chain could exhaust that limit on managers doing nothing but waiting their turn. A failed or cancelled manager simply never submits the next one, so the chain halts on its own — no separate cleanup needed for the not-yet-submitted remainder. refresh() picks up each newly chain-submitted job's ID automatically (written to a small chained_job_id.txt next to the group's logs) as soon as it's submitted.

  Manager job names are short and state which run(s) they own: i<NN>_mgr for a single-step manager (e.g. i04_mgr for run_04_...), i<NN>-<MM>_grp for a group manager spanning steps NN–MM (e.g. i02-03_grp) — handy for reading squeue at a glance.

  cfg = ISCE2_S1_Config(
      workdir='/data/p100_f466',
      bbox=[33.0, 38.0, -120.0, -115.0],
      hpc_mode=True,
      max_concurrent_hpc=12,   # default; tune to your cluster's fair-share limit
  )
  processor = Processor.create('ISCE2_S1', pairs=pairs, config=cfg)
  processor.submit()
  

  retry() auto-detects HPC mode from saved job metadata (slurm_job_ids / hpc_manager / hpc_array) — passing hpc_mode=True again is not required.

• Dry run

  Preview the run scripts and path checks without executing anything.

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

• Refresh

  Read step and command statuses from disk.

  jobs = processor.refresh()
  

• Retry failed steps

  Re-run all steps that have FAILED status.

  processor.retry()
  

• Cancel

  Terminate the running background process (local mode) or scancel all active SLURM jobs (HPC mode).

  processor.cancel()
  

• Watch

  Poll step statuses at regular intervals until all steps complete.

  processor.watch(refresh_interval=60)
  

• Save / Load

  Job state is saved automatically after submit(). To reload and resume from a saved job file:

  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()   # or .retry(), .cancel(), .watch()
  

• Running without a local ISCE2 install

  Set the container field to a path to an Apptainer/Singularity .sif image, or a Docker image reference (name[:tag]), and submit()/retry()/refresh()/watch()/cancel() all re-invoke the same insarhub processor ... CLI call inside that container instead of on the host — the workdir is bind-mounted at the identical path, so output lands exactly where a native run would put it, and ISCE2 never needs to be discovered on the host at all. The container image just needs insarhub installed alongside ISCE2/topsStack (see docker/dev/Dockerfile.isce2-mintpy for a ready-to-build example). The host running the app/CLI needs a container runtime (docker, or apptainer/singularity for a .sif) on its PATH; the container image itself does not need one — the pipeline runs directly inside it, never nesting another 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()
  

  The CLI form is the same:

  insarhub processor -N ISCE2_S1 -w /data/p100_f466 submit \\
      --container ghcr.io/jldz9/insarhub-isce2-mintpy:0.4.0
  

  container is persisted to the workdir's insarhub_config.json, so a later retry()/refresh()/cancel() (and, from the GUI, a retry) re-runs inside the same image without re-passing it. An explicit --container / container= on a later call overrides the saved value; a bare --container (no value) resolves to the processor's container_default image. container_default is a fixed per-processor suggestion (the image the GUI's "Run in Container" checkbox pre-fills) and is never persisted — only your actual container choice is. In HPC mode only each stage's child jobs run inside the container; the sbatch manager scaffolding stays on the host.

Runs GMTSAR's Python pipeline locally to build Sentinel-1 interferograms from .SAFE SLCs. The entry point is chosen by subswath:

• one IW (e.g. 2) — single-subswath, via p2p_processing
• several (e.g. "1 2 3", the default) — multi-subswath merged, via p2p_S1_TOPS_Frame

Either way callers pass raw .SAFE/.EOF names; the subswath and polarization are extracted internally.

GMTSAR runs in its own conda environment. gmtsar_root and gmtsar_env_bin locate it and both auto-detect, so pass them only when detection fails. Alternatively set container to a .sif/Docker image carrying insarhub+GMTSAR and skip local discovery.

• Import processor

  from insarhub import Processor
  

• Create 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-format DEM; auto-downloaded at staging if unset
      subswath      = 2,                 # IW2 only -- single-subswath. "1 2 3" (default) = multi-subswath merged
      gmtsar_root   = '/path/to/gmtsar',    # optional -- auto-detected if unset
      gmtsar_env_bin= '/path/to/conda/envs/gmtsar/bin',  # optional -- auto-detected if unset
  )
  pairs = [
      ("REF.SAFE", "REF.EOF", "SEC.SAFE", "SEC.EOF"),
  ]
  processor = Processor.create('GMTSAR_S1', pairs=pairs, config=cfg)
  

  Set dem_path for multi-subswath runs

  Multi-subswath gives each pair its own case directory. With dem_path unset the DEM is auto-downloaded at staging time — once per pair. On a 27-pair network that is the same DEM fetched 27 times.

  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.

• Submit

  Stage the GMTSAR case directory (and, for single-subswath mode, extract each pair's subswath), then launch p2p_processing/p2p_S1_TOPS_Frame in the background, up to max_workers concurrent pairs. Returns immediately; use refresh()/watch() to monitor progress.

  jobs = processor.submit()
  

• Submit (HPC / SLURM mode)

  Both modes support hpc_mode=True.

  p2p mode (stack_mode=False, the default) is the simpler of the two: every pair is completely independent — multi-subswath gives each its own case directory, and single-subswath output is namespaced intf/<julian_pair>/ — so a single sliding-window manager fans every pair out at once, max_concurrent_hpc live at a time, with no chaining at all. Each child job runs that pair's whole chain (align → interferogram → filter → unwrap → geocode) via the internal run-stage-unit --stage pair --index N re-entry. Job names are 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 instead runs each stack stage (align_F<N>/intf_F<N>/merge, or the flat align/intf for a single subswath) as its own sliding-window SLURM manager, instead of _run_stack() running as a background thread in the submitting process. Same chain-submission design as ISCE2_S1 (see its HPC-mode docs above): only the first stage's manager is submitted directly, and each chain-submits the next stage's manager itself right after it succeeds — never a --dependency chain pre-submitted up front — so at most one manager (plus its own ≤max_concurrent_hpc children) is ever sitting in the queue at a time. Manager job names are g_<stage>_mgr / children g_<stage>_<idx> (e.g. g_intf_F2_0007), the same short/self-describing convention as ISCE's i<NN>_mgr.

  One real difference from ISCE2_S1: GMTSAR_S1 has no flat shell-command-list generator the way stackSentinel.py's run_NN_* files give ISCE — each stage's real work lives in Python methods (_run_align_unit/_run_intf_unit/_run_merge_unit), so every HPC child job's "command" re-enters insarhub itself (the internal run-stage-unit CLI action) to call one of those methods in a fresh process, rather than a raw shell command line calling a GMTSAR binary directly.

  cfg = GMTSAR_S1_Config(
      workdir='/data/p100_f466',
      stack_mode=True,
      hpc_mode=True,
      max_concurrent_hpc=12,   # default; tune to your cluster's fair-share limit
      gmtsar_root=..., gmtsar_env_bin=...,
  )
  processor = Processor.create('GMTSAR_S1', pairs=pairs, config=cfg)
  processor.submit()
  

• Refresh

  Read per-pair status from GMTSAR's own output markers (.succeeded/.failed under intf/<julian_date_pair>/ (GMTSAR-assigned) or merge/).

  jobs = processor.refresh()
  

• Retry failed pairs

  Re-run only the pairs whose status is FAILED.

  processor.retry()
  

• Watch

  Poll pair statuses at regular intervals until all pairs reach SUCCEEDED or FAILED.

  processor.watch(poll_interval=60)
  

• Cancel (HPC mode)

  scancel every SLURM job (managers + their children) for an HPC submission, in either mode. p2p jobs are found from the hpc/p2p/ directory rather than config.hpc_mode, so a bare cancel locates them without repeating --hpc-mode; any pair still PENDING/RUNNING is marked FAILED so refresh does not report it as in flight. Local (non-HPC) stack_mode runs have nothing to cancel from a separate CLI invocation — _run_stack() runs as a background thread inside whichever process called submit(), not a detached background process the way ISCE2_S1's local mode is, so there's no separate process left running once that call returns.

  processor.cancel()
  

• Save

  Job state is saved automatically after submit() to <workdir>/gmtsar/gmtsar_jobs.json.

  processor.save()
  

• Output layout

  Single-subswath: <workdir>/gmtsar/intf/<julian_date_pair>/ (e.g. intf/2019184_2019196/ — GMTSAR's own Julian-date pair naming, not ref/sec stems) — GMTSAR's native file names (corr_ll.grd, phasefilt_ll.grd, *.PRM files), which is exactly what MintPy's prep_gmtsar.py expects directly.

  Multi-subswath: <workdir>/gmtsar/<ref_safe>_<sec_safe>/merge/ — the merged, geocoded product across every subswath named (phasefilt_ll.grd, corr_ll.grd, plus PNG/KML previews).

• Time series: use MintPy, not GMTSAR's own sbas

  This is a direct consequence of choosing p2p. GMTSAR's sbas works in radar coordinates and needs every SLC resampled onto one common grid — which per-pair alignment does not provide. MintPy's prep_gmtsar reads the geocoded *_ll.grd, so all pairs already share a geographic grid and no common alignment reference is required. Use the GMTSAR_Mintpy_SBAS analyzer.

• Running without a local GMTSAR install

  As with ISCE2_S1, set container to a .sif or Docker image carrying insarhub + GMTSAR and local discovery is skipped entirely:

  insarhub processor -N GMTSAR_S1 -w /data/stack submit \\
      --container ghcr.io/jldz9/insarhub-gmtsar-mintpy:0.4.0
  

  In HPC mode only each stage's child jobs run inside the container; the sbatch manager scaffolding stays on the host.

Builds an interferogram stack from ASF SLC-BURST granules using ISCE3/COMPASS for geocoding and dolphin downstream. Pair it with the S1_Burst downloader.

There is no coregistration: COMPASS geocodes every acquisition independently onto absolute UTM, so two dates of the same burst are pixel-aligned by construction.

Nine stages, run in order:

stage	tool	output
dem	sardem	Copernicus DEM + NASADEM water mask
tec	COMPASS	one IONEX map per acquisition date
cslc	s1_geocode_stack.py → run_*.sh	geocoded CSLC per burst-date
static	s1_static_layers.py	LOS/incidence geometry, then mosaicked onto the stack grid
crop	dolphin	each burst cut to the AOI
ifg	dolphin	interferograms (see ifg_mode)
stitch	dolphin	each pair's bursts merged into one raster
filt	dolphin	multilook → Goldstein → coherence
unwrap	snaphu	unwrapped phase + connected components

Choosing the estimator — ifg_mode

value	pairs come from	notes
phase_link (default)	full-covariance estimation	every pair contributes; pl_* fields tune it
network	a rule	n_connections, or max_temporal_baseline
user_defined	this folder's stack_*.json	exactly the pairs select_pairs chose

phase_link is the default because it measurably outperforms a pairwise network: on a test stack it gave one connected component at 83% coverage against three at 55%, and cut closure error from 0.157 to 0.067 rad. Its parameters match dolphin's own shipped configuration (glrt / 0.001, half-window 7×14, ministack 15).

Under phase_link + pl_ifg_network=single_reference (the defaults) a user-defined network is ignored — the estimator's output already is that network. Set pl_ifg_network=bandwidth if you want your pairs honoured there.

Processing extent

AOI is seeded automatically from the folder's downloader intersectsWith, so it is normally already filled in. Tick process_full_extent to process the whole downloaded burst footprint instead. Note crop_buffer_deg (0.05° by default) is added on every side — on a small AOI that buffer can approach the burst footprint on its own, so lower it if you want the AOI to actually bite.

dem and cslc run before anything is geocoded, so they always use AOI; process_full_extent applies from crop onward.

Notes

• Stages are decomposed for SLURM — see HPC (SLURM). cslc is one job per burst-date and dominates runtime; ifg under phase_link is a single job, because the estimator has no per-pair unit.
• The interferogram network is the intersection of dates across bursts. Where ASF has no coverage for one burst on one day, that date is excluded and named, so every pair stays formable on every burst.
• Time series is via the ISCE3_Dolphin_S1_PL analyzer, which serves both estimator modes.

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

Builds an interferogram stack from NISAR L2 GSLC granules using dolphin for phase-linking, interferograms and unwrapping. Pair it with the NISAR_GSLC downloader and the ISCE3_Dolphin_NISAR_PL analyzer.

Same dolphin engine as ISCE3_Burst but no geocoding — a GSLC is already geocoded, one frame per date, so the dem/tec/cslc/static stages are dropped and ifg is a single wrapped_phase.run over the whole stack.

Three stages, run in order:

stage	tool	output
ifg	dolphin	PS + phase-link + interferograms (one wrapped_phase.run over the stack)
stitch	dolphin	correlation + mosaic (trivial for one frame per date)
unwrap	snaphu	unwrapped phase + connected components

AOI cropping — why it matters here

A NISAR GSLC frame is enormous (e.g. 69840 × 68688 pixels), but the AOI is usually a small window. ISCE3_Burst gets its AOI cut for free from COMPASS's geocode --bbox; ISCE3_NISAR has no geocode step to do that, so it does the cut itself: at the ifg stage each GSLC's complex-SLC subdataset is windowed to the AOI into a lightweight VRT (gdal_translate -of VRT -projwin, cached in workdir/cropped_gslc/), and dolphin phase-links only that window.

Without the crop, dolphin would phase-link the entire frame and the stitch gdal_merge step would run out of memory on the full-frame rasters (the AOI would otherwise only be applied as a final output clip — too late to save the intermediate memory). On a real stack the crop reduced each input from 69840 × 68688 to ~8953 × 8728 (~60× smaller), turning a ~12 GB memory spike into a few hundred MB. Tick process_full_extent to disable the crop and process the whole geocoded frame instead (needs a large-memory host).

Configuration

• nisar_frequency (A default / B) and nisar_polarization (HH default, HV, VV, VH) select which GSLC grid group dolphin reads — /science/LSAR/GSLC/grids/frequency<freq>/<pol>. Keep these constant across a stack.
• AOI is seeded from the folder's downloader intersectsWith, same as ISCE3_Burst; the crop uses it directly.
• The pl_*, n_connections/max_temporal_baseline, and unwrap_* fields tune phase-linking, the interferogram network, and snaphu exactly as they do for ISCE3_Burst.

Notes

• The GSLC download itself is a full geocoded frame — the crop happens at processing time, not download time, so the .h5 products in slc/ are untouched and re-usable across AOIs.
• Runs on the insarhub-isce3-dolphin image (same as ISCE3_Burst / Dolphin_SBAS); set container to run without a local ISCE3/dolphin install, exactly as for the other processors.

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)