Skip to content

unwrapping

Stitch burst interferograms (optional) and unwrap them.

run(ifg_file_list, cor_file_list, nlooks, unwrap_options, temporal_coherence_filename=None, similarity_filename=None, mask_file=None, add_overviews=True)

Run the displacement workflow on a stack of SLCs.

Parameters:

Name Type Description Default
ifg_file_list Sequence[Path]

Sequence interferograms files to unwrap.

 required
cor_file_list Sequence[Path]

Sequence interferometric correlation files, one per file in ifg_file_list

 required
nlooks float

Effective number of looks used to form the input correlation data.

 required
unwrap_options UnwrapOptions

UnwrapOptions config object with parameters for running unwrapping jobs.

 required
temporal_coherence_filename Filename

Path to temporal coherence file from phase linking.

 None
similarity_filename Filename

Path to phase cosine similarity file from phase linking.

 None
mask_file PathOrStr

Path to boolean mask indicating nodata areas. 1 indicates valid data, 0 indicates missing data.

 None
add_overviews bool

If True, creates overviews of the unwrapped phase and connected component labels.

 = True

Returns:

Name Type Description
unwrapped_paths list[Path]

list of Paths to unwrapped interferograms created.
conncomp_paths list[Path]

list of Paths to connected component files created.
Source code in src/dolphin/workflows/unwrapping.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@log_runtime
def run(
    ifg_file_list: Sequence[Path],
    cor_file_list: Sequence[Path],
    nlooks: float,
    unwrap_options: UnwrapOptions,
    temporal_coherence_filename: Path | str | None = None,
    similarity_filename: Path | str | None = None,
    mask_file: Path | str | None = None,
    add_overviews: bool = True,
) -> tuple[list[Path], list[Path]]:
    """Run the displacement workflow on a stack of SLCs.

    Parameters
    ----------
    ifg_file_list : Sequence[Path]
        Sequence interferograms files to unwrap.
    cor_file_list : Sequence[Path]
        Sequence interferometric correlation files, one per file in `ifg_file_list`
    nlooks : float
        Effective number of looks used to form the input correlation data.
    unwrap_options : UnwrapOptions
        [`UnwrapOptions`][dolphin.workflows.config.UnwrapOptions] config object
        with parameters for running unwrapping jobs.
    temporal_coherence_filename : Filename, optional
        Path to temporal coherence file from phase linking.
    similarity_filename : Filename, optional
        Path to phase cosine similarity file from phase linking.
    mask_file : PathOrStr, optional
        Path to boolean mask indicating nodata areas.
        1 indicates valid data, 0 indicates missing data.
    add_overviews : bool, default = True
        If True, creates overviews of the unwrapped phase and connected component
        labels.

    Returns
    -------
    unwrapped_paths : list[Path]
        list of Paths to unwrapped interferograms created.
    conncomp_paths : list[Path]
        list of Paths to connected component files created.

    """
    t0 = time.perf_counter()
    if len(ifg_file_list) != len(cor_file_list):
        msg = f"{len(ifg_file_list) = } != {len(cor_file_list) = }"
        raise ValueError(msg)

    output_path = unwrap_options._directory
    output_path.mkdir(exist_ok=True, parents=True)
    if mask_file is not None:
        output_mask = _get_matching_raster(
            input_file=mask_file,
            output_dir=output_path,
            match_file=ifg_file_list[0],
        )
    else:
        output_mask = None

    logger.info(f"Unwrapping {len(ifg_file_list)} interferograms")

    # Make a scratch directory for unwrapping
    unwrap_scratchdir = unwrap_options._directory / "scratch"
    unwrap_scratchdir.mkdir(exist_ok=True, parents=True)

    unwrapped_paths, conncomp_paths = unwrap.run(
        ifg_filenames=ifg_file_list,
        cor_filenames=cor_file_list,
        output_path=output_path,
        unwrap_options=unwrap_options,
        nlooks=nlooks,
        temporal_coherence_filename=temporal_coherence_filename,
        similarity_filename=similarity_filename,
        mask_filename=output_mask,
        scratchdir=unwrap_scratchdir,
    )

    if add_overviews:
        logger.info("Creating overviews for unwrapped images")
        create_overviews(unwrapped_paths, image_type=ImageType.UNWRAPPED)
        create_overviews(conncomp_paths, image_type=ImageType.CONNCOMP)

    # Dump the used options for JSON parsing
    logger.info(
        "unwrapping complete",
        extra={
            "elapsed": time.perf_counter() - t0,
            "unwrap_options": unwrap_options.model_dump(mode="json"),
        },
    )

    return (unwrapped_paths, conncomp_paths)