Processor

The InSARHub Processor module provides functionality specifically for interferogram processing.

Import processor

Import the Processor class to access all dprocessor functionality
```
from insarhub import Processor
```
View available processors

List all registered processor
```
Processor.available()
```

Available Processors

Hyp3_InSAR

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_InSAR is specificially wrap insar_job in hype_sdk to provide InSAR SLC processing workflows.

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

class Hyp3_InSAR(Hyp3Base):
    name = "Hyp3_InSAR"
    default_config = Hyp3_InSAR_Config
    def __init__(self, config: Hyp3_InSAR_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_InSAR', workdir='/your/work/path', pairs=pairs)

OR

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

OR

from insarhub.config import Hyp3_InSAR_Config
cfg = Hyp3_InSAR_Config(workdir='/your/work/path', pairs=pairs)
processor = Processor.create('Hyp3_InSAR', 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_InSAR_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.

    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.
    """

    name: str = "Hyp3_InSAR_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
    phase_filter_parameter :float=0.6

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 Sucessed 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 previously saved json file and resume the work.
```
processor = Processor.create('Hyp3_InSAR', saved_job_path='path/to/your/json/file.json')
```
when load back saved job, you can resume the work to check/download jobs submitted to Hyp3 Server.

Processor

Available Processors

Hyp3_InSAR

Usage

⚠️ Major Redesign