config

Configuration and validation models for (sub)workflows.

CorrectionOptions

Bases: BaseModel

Configuration for the auxiliary phase corrections.

Source code in src/dolphin/workflows/config/_displacement.py

class CorrectionOptions(BaseModel, extra="forbid"):
    """Configuration for the auxiliary phase corrections."""

    _atm_directory: Path = Path("atmosphere")
    _iono_date_fmt: list[str] = ["%j0.%y", "%Y%j0000"]

    ionosphere_files: list[Path] = Field(
        default_factory=list,
        description=(
            "List of GNSS-derived TEC maps for ionospheric corrections (one per date)."
            " Source is https://cddis.nasa.gov/archive/gnss/products/ionex/"
        ),
    )

    geometry_files: list[Path] = Field(
        default_factory=list,
        description=(
            "Line-of-sight geometry files for each burst/SLC stack area, for use in"
            " correction computations."
        ),
    )
    dem_file: Optional[Path] = Field(
        None,
        description="DEM file for tropospheric/ topographic phase corrections.",
    )

    @field_validator("ionosphere_files", "geometry_files", mode="before")
    @classmethod
    def _to_empty_list(cls, v):
        return v if v is not None else []

DisplacementWorkflow

Bases: WorkflowBase

Configuration for the workflow.

Source code in src/dolphin/workflows/config/_displacement.py

class DisplacementWorkflow(WorkflowBase):
    """Configuration for the workflow."""

    # Paths to input/output files
    input_options: InputOptions = Field(default_factory=InputOptions)
    cslc_file_list: Annotated[
        list[Path],
        # Add aliases for the CLI
        tyro.conf.arg(aliases=("--cslc", "--slc-files")),
    ] = Field(
        default_factory=list,
        description=(
            "list of CSLC files, or newline-delimited file "
            "containing list of CSLC files."
        ),
    )
    require_cslc_files: bool = Field(True, description="", exclude=True)

    output_options: OutputOptions = Field(default_factory=OutputOptions)

    # Options for each step in the workflow
    ps_options: PsOptions = Field(default_factory=PsOptions)
    amplitude_dispersion_files: list[Path] = Field(
        default_factory=list,
        description=(
            "Paths to existing Amplitude Dispersion file (1 per SLC region) for PS"
            " update calculation. If none provided, computed using the input SLC stack."
        ),
    )
    amplitude_mean_files: list[Path] = Field(
        default_factory=list,
        description=(
            "Paths to an existing Amplitude Mean files (1 per SLC region) for PS update"
            " calculation. If none provided, computed using the input SLC stack."
        ),
    )
    layover_shadow_mask_files: list[Path] = Field(
        default_factory=list,
        description=(
            "Paths to layover/shadow binary masks, where 0 indicates a pixel in"
            " layover/shadow, 1 is a good pixel. If none provided, no masking is"
            " performed for layover/shadow."
        ),
    )

    phase_linking: PhaseLinkingOptions = Field(default_factory=PhaseLinkingOptions)
    interferogram_network: InterferogramNetwork = Field(
        default_factory=InterferogramNetwork
    )
    unwrap_options: UnwrapOptions = Field(default_factory=UnwrapOptions)
    timeseries_options: TimeseriesOptions = Field(default_factory=TimeseriesOptions)

    # internal helpers
    # Stores the list of directories to be created by the workflow
    model_config = ConfigDict(
        extra="allow", json_schema_extra={"required": ["cslc_file_list"]}
    )

    # validators
    # reuse the _read_file_list_or_glob
    _check_cslc_file_glob = field_validator("cslc_file_list", mode="before")(
        _read_file_list_or_glob
    )

    @model_validator(mode="after")
    def _check_zero_interferogram_network(self: Self) -> Self:
        ifg_network = self.interferogram_network
        ref_idx = ifg_network.reference_idx
        max_bw = ifg_network.max_bandwidth
        max_tb = ifg_network.max_temporal_baseline
        indexes = ifg_network.indexes
        # Check if more than one has been set:
        if ref_idx is None and max_bw is None and max_tb is None and indexes is None:
            logger.info(
                "No network configuration options were set. Using Nearest-3 network"
            )
            self.interferogram_network.max_bandwidth = 3
        return self

    @model_validator(mode="after")
    def _check_input_files_exist(self) -> Self:
        if not self.require_cslc_files:
            return self
        file_list = self.cslc_file_list
        if not file_list:
            msg = "Must specify list of input SLC files."
            raise ValueError(msg)

        input_options = self.input_options
        date_fmt = input_options.cslc_date_fmt
        # Filter out files that don't have dates in the filename
        files_matching_date = [Path(f) for f in file_list if get_dates(f, fmt=date_fmt)]
        if len(files_matching_date) < len(file_list):
            msg = (
                f"Found {len(files_matching_date)} files with dates like {date_fmt} in"
                f" the filename out of {len(file_list)} files."
            )
            raise ValueError(msg)

        ext = file_list[0].suffix
        # If they're HDF5/NetCDF files, we need to check that the subdataset exists
        if ext in [".h5", ".nc"]:
            subdataset = input_options.subdataset
            if subdataset is None:
                msg = "Must provide subdataset name for input NetCDF/HDF5 files."
                raise ValueError(msg)

        # Coerce the file_list to a sorted list of Path objects
        self.cslc_file_list = [
            Path(f) for f in sort_files_by_date(file_list, file_date_fmt=date_fmt)[0]
        ]

        return self

    def model_post_init(self, context: Any, /) -> None:
        """After validation, set up properties for use during workflow run."""
        super().model_post_init(context)

        if self.input_options.wavelength is None and self.cslc_file_list:
            # Try to infer the wavelength from filenames
            try:
                get_burst_id(self.cslc_file_list[-1])
                # The Burst ID was recognized for OPERA-S1 SLCs: use S1 wavelength
                self.input_options.wavelength = constants.SENTINEL_1_WAVELENGTH
            except ValueError:
                pass

        # Ensure outputs from workflow steps are within work directory.
        if not self.keep_paths_relative:
            # Resolve all CSLC paths (skip remote URLs like https://, s3://):
            self.cslc_file_list = [
                p if is_remote_url(p) else p.resolve(strict=False)
                for p in self.cslc_file_list
            ]

        work_dir = self.work_directory
        # For each workflow step that has an output folder, move it inside
        # the work directory (if it's not already inside).
        # They may already be inside if we're loading from a json/yaml file.
        for step in [
            "ps_options",
            "phase_linking",
            "interferogram_network",
            "unwrap_options",
            "timeseries_options",
        ]:
            opts = getattr(self, step)
            if isinstance(opts, dict):
                # If this occurs, we are printing the schema.
                # Using newer pydantic `model_construct`, this would be a dict,
                # instead of an object.
                # We don't care about the subsequent logic here
                return

            if opts._directory.parent != work_dir:
                opts._directory = work_dir / opts._directory
            if not self.keep_paths_relative:
                opts._directory = opts._directory.resolve(strict=False)

        # Track the directories that need to be created at start of workflow
        self._directory_list = [
            work_dir,
            self.ps_options._directory,
            self.phase_linking._directory,
            self.interferogram_network._directory,
            self.unwrap_options._directory,
            self.timeseries_options._directory,
        ]
        # Add the output PS files we'll create to the `PS` directory, making
        # sure they're inside the work directory
        ps_opts = self.ps_options
        ps_opts._amp_dispersion_file = work_dir / ps_opts._amp_dispersion_file
        ps_opts._amp_mean_file = work_dir / ps_opts._amp_mean_file
        ps_opts._output_file = work_dir / ps_opts._output_file

        self.timeseries_options._velocity_file = (
            work_dir / self.timeseries_options._velocity_file
        )

        # Modify interferogram options if using spurt for 3d unwrapping,
        # which only does nearest-3 interferograms
        if self.unwrap_options.unwrap_method == UnwrapMethod.SPURT:
            logger.info(
                "Using spurt: will form single reference interferograms, later convert"
                " to nearest-3"
            )
            self.interferogram_network.reference_idx = 0
            # Force all other network options to None
            for attr in ["max_bandwidth", "max_temporal_baseline", "indexes"]:
                setattr(self.interferogram_network, attr, None)

model_post_init(context)

After validation, set up properties for use during workflow run.

Source code in src/dolphin/workflows/config/_displacement.py

def model_post_init(self, context: Any, /) -> None:
    """After validation, set up properties for use during workflow run."""
    super().model_post_init(context)

    if self.input_options.wavelength is None and self.cslc_file_list:
        # Try to infer the wavelength from filenames
        try:
            get_burst_id(self.cslc_file_list[-1])
            # The Burst ID was recognized for OPERA-S1 SLCs: use S1 wavelength
            self.input_options.wavelength = constants.SENTINEL_1_WAVELENGTH
        except ValueError:
            pass

    # Ensure outputs from workflow steps are within work directory.
    if not self.keep_paths_relative:
        # Resolve all CSLC paths (skip remote URLs like https://, s3://):
        self.cslc_file_list = [
            p if is_remote_url(p) else p.resolve(strict=False)
            for p in self.cslc_file_list
        ]

    work_dir = self.work_directory
    # For each workflow step that has an output folder, move it inside
    # the work directory (if it's not already inside).
    # They may already be inside if we're loading from a json/yaml file.
    for step in [
        "ps_options",
        "phase_linking",
        "interferogram_network",
        "unwrap_options",
        "timeseries_options",
    ]:
        opts = getattr(self, step)
        if isinstance(opts, dict):
            # If this occurs, we are printing the schema.
            # Using newer pydantic `model_construct`, this would be a dict,
            # instead of an object.
            # We don't care about the subsequent logic here
            return

        if opts._directory.parent != work_dir:
            opts._directory = work_dir / opts._directory
        if not self.keep_paths_relative:
            opts._directory = opts._directory.resolve(strict=False)

    # Track the directories that need to be created at start of workflow
    self._directory_list = [
        work_dir,
        self.ps_options._directory,
        self.phase_linking._directory,
        self.interferogram_network._directory,
        self.unwrap_options._directory,
        self.timeseries_options._directory,
    ]
    # Add the output PS files we'll create to the `PS` directory, making
    # sure they're inside the work directory
    ps_opts = self.ps_options
    ps_opts._amp_dispersion_file = work_dir / ps_opts._amp_dispersion_file
    ps_opts._amp_mean_file = work_dir / ps_opts._amp_mean_file
    ps_opts._output_file = work_dir / ps_opts._output_file

    self.timeseries_options._velocity_file = (
        work_dir / self.timeseries_options._velocity_file
    )

    # Modify interferogram options if using spurt for 3d unwrapping,
    # which only does nearest-3 interferograms
    if self.unwrap_options.unwrap_method == UnwrapMethod.SPURT:
        logger.info(
            "Using spurt: will form single reference interferograms, later convert"
            " to nearest-3"
        )
        self.interferogram_network.reference_idx = 0
        # Force all other network options to None
        for attr in ["max_bandwidth", "max_temporal_baseline", "indexes"]:
            setattr(self.interferogram_network, attr, None)

HalfWindow

Bases: BaseModel

Class to hold half-window size for multi-looking during phase linking.

Source code in src/dolphin/workflows/config/_common.py

class HalfWindow(BaseModel, extra="forbid"):
    """Class to hold half-window size for multi-looking during phase linking."""

    y: Annotated[
        int,
        tyro.conf.arg(aliases=("--hwy",)),
    ] = Field(7, description="Half window size (in pixels) for y direction", gt=0)
    x: Annotated[
        int,
        tyro.conf.arg(aliases=("--hwx",)),
    ] = Field(14, description="Half window size (in pixels) for x direction", gt=0)

    def to_looks(self):
        """Convert (x, y) half-window size to (row, column) looks."""
        return 2 * self.y + 1, 2 * self.x + 1

    @classmethod
    def from_looks(cls, row_looks: int, col_looks: int):
        """Create a half-window from looks."""
        return cls(x=col_looks // 2, y=row_looks // 2)

from_looks(row_looks, col_looks) classmethod

Create a half-window from looks.

Source code in src/dolphin/workflows/config/_common.py

@classmethod
def from_looks(cls, row_looks: int, col_looks: int):
    """Create a half-window from looks."""
    return cls(x=col_looks // 2, y=row_looks // 2)

to_looks()

Convert (x, y) half-window size to (row, column) looks.

Source code in src/dolphin/workflows/config/_common.py

def to_looks(self):
    """Convert (x, y) half-window size to (row, column) looks."""
    return 2 * self.y + 1, 2 * self.x + 1

InputOptions

Bases: BaseModel

Options specifying input datasets for workflow.

Source code in src/dolphin/workflows/config/_common.py

class InputOptions(BaseModel, extra="forbid"):
    """Options specifying input datasets for workflow."""

    subdataset: Annotated[
        Optional[str],
        tyro.conf.arg(aliases=("--subdataset", "--sds")),
    ] = Field(
        None,
        description="If passing HDF5/NetCDF files, subdataset to use from CSLC files. ",
    )
    cslc_date_fmt: str = Field(
        "%Y%m%d",
        description="Format of dates contained in CSLC filenames",
    )
    wavelength: Optional[float] = Field(
        None,
        description=(
            "Radar wavelength (in meters) of the transmitted data. used to convert the"
            " units in the rasters in `timeseries/` to from radians to meters. If None"
            " and sensor is not recognized, outputs remain in radians."
        ),
    )
    azimuth_blocks: int = Field(
        1,
        description=(
            "When the input does not match OPERA-burst naming (e.g. NISAR), split"
            " each input frame into this many azimuth blocks and process each block"
            " as a synthetic burst. Default 1 = no splitting."
        ),
        ge=1,
    )
    halo_rows: Optional[int] = Field(
        None,
        description=(
            "Halo (input rows) on each side of an azimuth block. Default:"
            " max(half_window_y, similarity_search_radius * stride_y,"
            " (corr_window_y // 2) * stride_y) + 5."
        ),
    )

InterferogramNetwork

Bases: BaseModel

Options to determine the type of network for interferogram formation.

Source code in src/dolphin/workflows/config/_common.py

class InterferogramNetwork(BaseModel, extra="forbid"):
    """Options to determine the type of network for interferogram formation."""

    _directory: Path = PrivateAttr(Path("interferograms"))

    reference_idx: Optional[int] = Field(
        None,
        description=(
            "For single-reference network: Index of the reference image in the network"
        ),
    )
    max_bandwidth: Optional[int] = Field(
        None,
        description="Max `n` to form the nearest-`n` interferograms by index.",
        ge=1,
    )
    max_temporal_baseline: Optional[int] = Field(
        None,
        description="Maximum temporal baseline of interferograms.",
        ge=0,
    )
    indexes: Optional[list[tuple[int, int]]] = Field(
        None,
        description=(
            "For manual-index network: list of (ref_idx, sec_idx) defining the"
            " interferograms to form."
        ),
    )

OutputOptions

Bases: BaseModel

Options for the output size/format/compressions.

Source code in src/dolphin/workflows/config/_common.py

class OutputOptions(BaseModel, extra="forbid"):
    """Options for the output size/format/compressions."""

    strides: Strides = Field(Strides(), validate_default=True)
    epsg: Optional[int] = Field(
        None,
        description=(
            "EPSG code of desired output products (for geocoded SLCs only)."
            " If None, uses the most common projection of the input data."
        ),
    )
    bounds: Optional[tuple[float, float, float, float]] = Field(
        None,
        description=(
            "Area of interest: [left, bottom, right, top] coordinates. "
            "e.g. `bbox=[-150.2,65.0,-150.1,65.5]`"
        ),
    )
    bounds_epsg: Optional[int] = Field(
        4326,
        description=(
            "EPSG code for the `bounds` or `bounds_wkt` coordinates, if specified."
        ),
    )
    bounds_wkt: Optional[str] = Field(
        None,
        description=(
            "Area of interest as a simple Polygon in well-known-text (WKT) format."
            " Can pass a string, or a `.wkt` filename containing the Polygon text."
        ),
    )

    hdf5_creation_options: dict = Field(
        DEFAULT_HDF5_OPTIONS,
        description="Options for `create_dataset` with h5py.",
    )
    gtiff_creation_options: list[str] = Field(
        list(DEFAULT_TIFF_OPTIONS),
        description="GDAL creation options for GeoTIFF files",
    )
    add_overviews: bool = Field(
        True,
        description=(
            "Whether to add overviews to the output GeoTIFF files. This will "
            "increase file size, but can be useful for visualizing the data with "
            "web mapping tools. See https://gdal.org/programs/gdaladdo.html for more."
        ),
    )
    overview_levels: list[int] = Field(
        [4, 8, 16, 32, 64],
        description="List of overview levels to create (if `add_overviews=True`).",
    )
    # Note: we use NaiveDatetime, since other datetime parsing results in Naive
    # (no TzInfo) datetimes, which can't be compared to datetimes with timezones
    extra_reference_date: Optional[datetime] = Field(
        None,
        description=(
            "Specify an extra reference datetime in UTC. Adding this lets you"
            " to create and unwrap two single reference networks; the later resets at"
            " the given date (e.g. for a large earthquake event). If passing strings,"
            " formats accepted are YYYY-MM-DD[T]HH:MM[:SS[.ffffff]][Z or [±]HH[:]MM],"
            " or YYYY-MM-DD"
        ),
    )

    # validators
    @field_validator("bounds_wkt", mode="after")
    @classmethod
    def _read_wkt_file(cls, bounds_wkt: str):
        if bounds_wkt and bounds_wkt.endswith(".wkt"):
            return Path(bounds_wkt).read_text()
        return bounds_wkt

    @field_validator("bounds", mode="after")
    @classmethod
    def _check_and_convert_bounds(cls, bounds, info):
        bounds_wkt = info.data.get("bounds_wkt")
        if bounds is not None and bounds_wkt is not None:
            msg = "Cannot specify both bounds and bounds_wkt."
            raise ValueError(msg)
        if bounds:
            return Bbox(*bounds)
        return bounds

    @field_validator("bounds_epsg", mode="after")
    @classmethod
    def _ensure_bounds_epsg(cls, bounds_epsg, info):
        bounds = info.data.get("bounds")
        if bounds is not None and bounds_epsg is None:
            msg = "Must specify `bounds_epsg` if `bounds` is provided."
            raise ValueError(msg)
        return bounds_epsg

    @field_validator("extra_reference_date", mode="after")
    @classmethod
    def _strip_timezone(cls, extra_reference_date):
        if extra_reference_date is None:
            return None
        return extra_reference_date.replace(tzinfo=None)

PhaseLinkingOptions

Bases: BaseModel

Configurable options for wrapped phase estimation.

Source code in src/dolphin/workflows/config/_common.py

class PhaseLinkingOptions(BaseModel, extra="forbid"):
    """Configurable options for wrapped phase estimation."""

    _directory: Path = PrivateAttr(Path("linked_phase"))
    ministack_size: Annotated[
        int,
        tyro.conf.arg(aliases=("--ms",)),
    ] = Field(15, description="Size of the ministack for sequential estimator.", gt=1)
    max_num_compressed: int = Field(
        10,
        description=(
            "Maximum number of compressed images to use in sequential estimator."
            " If there are more ministacks than this, the earliest CCSLCs will be"
            " left out of the later stacks. "
        ),
        gt=0,
    )
    output_reference_idx: Optional[int] = Field(
        None,
        description=(
            "Index of the SLC to use as interferogram reference after phase linking. If"
            " not set, uses the CompressedSlcPlan default"
        ),
    )
    half_window: HalfWindow = HalfWindow()
    use_evd: Annotated[
        bool,
        tyro.conf.arg(aliases=("--use-evd",)),
    ] = Field(
        False, description="Use EVD on the coherence instead of using the EMI algorithm"
    )

    beta: float = Field(
        0.00,
        description=(
            "Beta regularization parameter for correlation matrix inversion. 0 is no"
            " regularization."
        ),
        ge=0.0,
        le=1.0,
    )
    zero_correlation_threshold: float = Field(
        0.00,
        description=(
            "Snap correlation values in the coherence matrix below this value to 0."
        ),
        ge=0.0,
        le=1.0,
    )
    shp_method: ShpMethod = ShpMethod.GLRT
    shp_alpha: float = Field(
        0.001,
        description=(
            "Significance level (probability of false alarm) for SHP tests. Lower"
            " numbers include more pixels within the multilook window during covariance"
            " estimation."
        ),
        gt=0.0,
        lt=1.0,
    )
    mask_input_ps: bool = Field(
        False,
        description=(
            "If True, pixels labeled as PS will get set to NaN during phase linking to"
            " avoid summing their phase. Default of False means that the SHP algorithm"
            " will decide if a pixel should be included, regardless of its PS label."
        ),
    )
    baseline_lag: Optional[int] = Field(
        None,
        gt=0,
        description=(
            "StBAS parameter to include only nearest-N interferograms for"
            "phase linking. A `baseline_lag` of `n` will only include the closest"
            "`n` interferograms. `baseline_line` must be positive."
        ),
    )
    compressed_slc_plan: CompressedSlcPlan = CompressedSlcPlan.ALWAYS_FIRST
    write_crlb: bool = Field(
        True,
        description=(
            "Write out (and stitch, if processing multiple geocoded bursts) rasters"
            " containing the Cramer-Rao Lower Bound (CRLB) estimates."
        ),
    )
    write_closure_phase: bool = Field(
        False,
        description=(
            "Write out (and stitch, if processing multiple geocoded bursts) rasters"
            " containing the sequential closure phase from the nearest-3 triplets in"
            " the coherence matrix."
        ),
    )

PsOptions

Bases: BaseModel

Options for the PS pixel selection portion of the workflow.

Source code in src/dolphin/workflows/config/_common.py

class PsOptions(BaseModel, extra="forbid"):
    """Options for the PS pixel selection portion of the workflow."""

    _directory: Path = PrivateAttr(Path("PS"))
    _output_file: Path = PrivateAttr(Path("PS/ps_pixels.tif"))
    _amp_dispersion_file: Path = PrivateAttr(Path("PS/amp_dispersion.tif"))
    _amp_mean_file: Path = PrivateAttr(Path("PS/amp_mean.tif"))

    amp_dispersion_threshold: float = Field(
        0.25,
        description="Amplitude dispersion threshold to consider a pixel a PS.",
        ge=0.0,
    )

PsWorkflow

Bases: WorkflowBase

Configuration for the workflow.

Source code in src/dolphin/workflows/config/_ps.py

class PsWorkflow(WorkflowBase):
    """Configuration for the workflow."""

    # Paths to input/output files
    input_options: InputOptions = Field(default_factory=InputOptions)
    cslc_file_list: list[Path] = Field(
        default_factory=list,
        description=(
            "list of CSLC files, or newline-delimited file "
            "containing list of CSLC files."
        ),
    )

    # Options for each step in the workflow
    ps_options: PsOptions = Field(default_factory=PsOptions)
    output_options: OutputOptions = Field(default_factory=OutputOptions)
    layover_shadow_mask_files: list[Path] = Field(
        default_factory=list,
        description=(
            "Paths to layover/shadow binary masks, where 0 indicates a pixel in"
            " layover/shadow, 1 is a good pixel. If none provided, no masking is"
            " performed for layover/shadow."
        ),
    )

    # internal helpers
    model_config = ConfigDict(
        extra="allow", json_schema_extra={"required": ["cslc_file_list"]}
    )

    # validators
    # reuse the _read_file_list_or_glob
    _check_cslc_file_glob = field_validator("cslc_file_list", mode="before")(
        _read_file_list_or_glob
    )

    def model_post_init(self, context: Any, /) -> None:
        """After validation, set up properties for use during workflow run."""
        super().model_post_init(context)
        # Ensure outputs from workflow steps are within work directory.
        if not self.keep_paths_relative:
            # Resolve all CSLC paths (skip remote URLs like https://, s3://):
            self.cslc_file_list = [
                p if is_remote_url(p) else p.resolve(strict=False)
                for p in self.cslc_file_list
            ]

        work_dir = self.work_directory
        # move the folders inside the work directory
        ps_opts = self.ps_options
        if ps_opts._directory.parent != work_dir:
            ps_opts._directory = work_dir / ps_opts._directory
        if not self.keep_paths_relative:
            ps_opts._directory = ps_opts._directory.resolve(strict=False)

        # Track the directories that need to be created at start of workflow
        self._directory_list = [
            work_dir,
            self.ps_options._directory,
        ]
        # Add the output PS files we'll create to the `PS` directory, making
        # sure they're inside the work directory
        ps_opts._amp_dispersion_file = work_dir / ps_opts._amp_dispersion_file
        ps_opts._amp_mean_file = work_dir / ps_opts._amp_mean_file
        ps_opts._output_file = work_dir / ps_opts._output_file

model_post_init(context)

After validation, set up properties for use during workflow run.

Source code in src/dolphin/workflows/config/_ps.py

def model_post_init(self, context: Any, /) -> None:
    """After validation, set up properties for use during workflow run."""
    super().model_post_init(context)
    # Ensure outputs from workflow steps are within work directory.
    if not self.keep_paths_relative:
        # Resolve all CSLC paths (skip remote URLs like https://, s3://):
        self.cslc_file_list = [
            p if is_remote_url(p) else p.resolve(strict=False)
            for p in self.cslc_file_list
        ]

    work_dir = self.work_directory
    # move the folders inside the work directory
    ps_opts = self.ps_options
    if ps_opts._directory.parent != work_dir:
        ps_opts._directory = work_dir / ps_opts._directory
    if not self.keep_paths_relative:
        ps_opts._directory = ps_opts._directory.resolve(strict=False)

    # Track the directories that need to be created at start of workflow
    self._directory_list = [
        work_dir,
        self.ps_options._directory,
    ]
    # Add the output PS files we'll create to the `PS` directory, making
    # sure they're inside the work directory
    ps_opts._amp_dispersion_file = work_dir / ps_opts._amp_dispersion_file
    ps_opts._amp_mean_file = work_dir / ps_opts._amp_mean_file
    ps_opts._output_file = work_dir / ps_opts._output_file

ShpMethod

Bases: str, Enum

Method for finding SHPs during phase linking.

Source code in src/dolphin/workflows/config/_enums.py

class ShpMethod(str, Enum):
    """Method for finding SHPs during phase linking."""

    GLRT = "glrt"
    KS = "ks"
    RECT = "rect"
    # Alias for no SHP search
    NONE = "rect"

SpurtOptions

Bases: BaseModel

Options for running 3D unwrapping on a set of interferograms.

Uses spurt to run the temporal/spatial unwrapping. Options are passed through to spurt library.

Source code in src/dolphin/workflows/config/_unwrap_options.py

class SpurtOptions(BaseModel, extra="forbid"):
    """Options for running 3D unwrapping on a set of interferograms.

    Uses [`spurt`](https://github.com/isce-framework/spurt) to run the
    temporal/spatial unwrapping. Options are passed through to `spurt`
    library.

    """

    temporal_coherence_threshold: float = Field(
        0.7,
        description="Temporal coherence to pick pixels used on an irregular grid.",
        ge=0.0,
        lt=1.0,
    )
    similarity_threshold: float = Field(
        0.5,
        description=(
            "Similarity to pick pixels used on an irregular grid. Any pixel with"
            " similarity above `similarity_threshold` *or* above the temporal coherence"
            " threshold is chosen."
        ),
        ge=0.0,
        lt=1.0,
    )
    run_ambiguity_interpolation: bool = Field(
        True,
        description=(
            "After running spurt, interpolate the values that were masked during"
            " unwrapping (which are otherwise left as nan)."
        ),
    )
    general_settings: SpurtGeneralSettings = Field(default_factory=SpurtGeneralSettings)
    tiler_settings: SpurtTilerSettings = Field(default_factory=SpurtTilerSettings)
    solver_settings: SpurtSolverSettings = Field(default_factory=SpurtSolverSettings)
    merger_settings: SpurtMergerSettings = Field(default_factory=SpurtMergerSettings)

Strides

Bases: BaseModel

Specify the strides (decimation factor) to perform while processing input.

For example, strides of {x: 4, y: 2} would turn an input of shape (100, 100) into an output of shape (50, 25).

Source code in src/dolphin/workflows/config/_common.py

class Strides(BaseModel, extra="forbid"):
    """Specify the strides (decimation factor) to perform while processing input.

    For example, strides of {x: 4, y: 2} would turn an input of shape (100, 100)
    into an output of shape (50, 25).
    """

    y: Annotated[int, tyro.conf.arg(aliases=("--sy",))] = Field(
        1, description="Decimation factor (stride) in the y/row direction", gt=0
    )
    x: Annotated[int, tyro.conf.arg(aliases=("--sx",))] = Field(
        1, description="Decimation factor (stride) in the x/column direction", gt=0
    )

TimeseriesOptions

Bases: BaseModel

Options for inversion/time series fitting.

Source code in src/dolphin/workflows/config/_common.py

class TimeseriesOptions(BaseModel, extra="forbid"):
    """Options for inversion/time series fitting."""

    _directory: Path = PrivateAttr(Path("timeseries"))
    _velocity_file: Path = PrivateAttr(Path("timeseries/velocity.tif"))
    run_inversion: bool = Field(
        True,
        description=(
            "Whether to run the inversion step after unwrapping, if more than "
            " a single-reference network is used."
        ),
    )
    method: Literal["L1", "L2"] = Field(
        "L1", description="Norm to use during timeseries inversion."
    )
    reference_point: Optional[tuple[int, int]] = Field(
        None,
        description=(
            "Reference point (row, col) used if performing a time series inversion. "
            "If not provided, a point will be selected from a consistent connected "
            "component with low amplitude dispersion."
        ),
    )

    run_velocity: bool = Field(
        True,
        description="Run the velocity estimation from the phase time series.",
    )
    apply_mask_to_timeseries: bool = Field(
        True,
        description=(
            "If providing a mask file, whether to apply the mask to the timeseries. If"
            " True, the time series rasters will zero out pixels labeled as 0 in the"
            " mask file."
        ),
    )
    correlation_threshold: float = Field(
        0.2,
        description="Pixels with correlation below this value will be masked out.",
        ge=0.0,
        le=1.0,
    )
    block_shape: tuple[int, int] = Field(
        (256, 256),
        description=(
            "Size (rows, columns) of blocks of data to load at a time. 3D dimension is"
            " number of interferograms (during inversion) and number of SLC dates"
            " (during velocity fitting)"
        ),
    )
    num_parallel_blocks: int = Field(
        4, description="Number of parallel blocks to process at once."
    )

UnwrapMethod

Bases: str, Enum

Phase unwrapping method.

Source code in src/dolphin/workflows/config/_enums.py

class UnwrapMethod(str, Enum):
    """Phase unwrapping method."""

    SNAPHU = "snaphu"
    ICU = "icu"
    PHASS = "phass"
    SPURT = "spurt"
    WHIRLWIND = "whirlwind"

UnwrapOptions

Bases: BaseModel

Options for unwrapping after wrapped phase estimation.

Source code in src/dolphin/workflows/config/_unwrap_options.py

class UnwrapOptions(BaseModel, extra="forbid"):
    """Options for unwrapping after wrapped phase estimation."""

    run_unwrap: bool = Field(
        True,
        description=(
            "Whether to run the unwrapping step after wrapped phase estimation."
        ),
    )
    run_goldstein: bool = Field(
        False,
        description="Whether to run Goldstein filtering step on wrapped interferogram.",
    )
    run_interpolation: bool = Field(
        False,
        description="Whether to run interpolation step on wrapped interferogram.",
    )
    _directory: Path = PrivateAttr(Path("unwrapped"))
    unwrap_method: UnwrapMethod = UnwrapMethod.SNAPHU
    n_parallel_jobs: int = Field(
        -1,
        description=(
            "Number of interferograms to unwrap in parallel."
            " ``-1`` (default) auto-selects based on the unwrap method:"
            " ``1`` for snaphu/spurt (which already parallelise"
            " internally — snaphu over tiles, spurt over solver workers),"
            " and ``max(1, cpu_count // 4)`` for whirlwind (its MCF"
            " Amdahl-saturates around 4 threads per IG, so concurrency"
            " gives a ~2x wall-clock speedup at no per-IG cost)."
        ),
        ge=-1,
    )
    zero_where_masked: bool = Field(
        False,
        description=(
            "Set wrapped phase/correlation to 0 where mask is 0 before unwrapping. "
        ),
    )
    preprocess_options: PreprocessOptions = Field(default_factory=PreprocessOptions)
    snaphu_options: SnaphuOptions = Field(default_factory=SnaphuOptions)
    tophu_options: TophuOptions = Field(default_factory=TophuOptions)
    spurt_options: SpurtOptions = Field(default_factory=SpurtOptions)
    whirlwind_options: WhirlwindOptions = Field(default_factory=WhirlwindOptions)

UnwrappingWorkflow

Bases: WorkflowBase

Configuration for the unwrapping stage of the workflow.

Source code in src/dolphin/workflows/config/_unwrapping.py

class UnwrappingWorkflow(WorkflowBase):
    """Configuration for the unwrapping stage of the workflow."""

    # Paths to input/output files
    ifg_file_list: list[Path] = Field(
        default_factory=list,
        description=(
            "list of CSLC files, or newline-delimited file "
            "containing list of CSLC files."
        ),
    )
    input_options: InputOptions = Field(default_factory=InputOptions)
    output_options: OutputOptions = Field(default_factory=OutputOptions)

    unwrap_options: UnwrapOptions = Field(default_factory=UnwrapOptions)

    # internal helpers
    # Stores the list of directories to be created by the workflow
    model_config = ConfigDict(
        extra="allow", json_schema_extra={"required": ["ifg_file_list"]}
    )

    # validators
    # reuse the _read_file_list_or_glob
    _check_cslc_file_glob = field_validator("ifg_file_list", mode="before")(
        _read_file_list_or_glob
    )

    def model_post_init(self, context: Any, /) -> None:
        """After validation, set up properties for use during workflow run."""
        super().model_post_init(context)
        # Ensure outputs from workflow steps are within work directory.
        if not self.keep_paths_relative:
            # Resolve all CSLC paths:
            self.ifg_file_list = [p.resolve(strict=False) for p in self.ifg_file_list]

        work_dir = self.work_directory
        # move output dir inside the work directory (if it's not already inside).
        # They may already be inside if we're loading from a json/yaml file.
        opts = self.unwrap_options
        if opts._directory.parent != work_dir:
            opts._directory = work_dir / opts._directory
        if not self.keep_paths_relative:
            opts._directory = opts._directory.resolve(strict=False)

        # Track the directories that need to be created at start of workflow
        self._directory_list = [
            work_dir,
            self.unwrap_options._directory,
        ]

model_post_init(context)

After validation, set up properties for use during workflow run.

Source code in src/dolphin/workflows/config/_unwrapping.py

def model_post_init(self, context: Any, /) -> None:
    """After validation, set up properties for use during workflow run."""
    super().model_post_init(context)
    # Ensure outputs from workflow steps are within work directory.
    if not self.keep_paths_relative:
        # Resolve all CSLC paths:
        self.ifg_file_list = [p.resolve(strict=False) for p in self.ifg_file_list]

    work_dir = self.work_directory
    # move output dir inside the work directory (if it's not already inside).
    # They may already be inside if we're loading from a json/yaml file.
    opts = self.unwrap_options
    if opts._directory.parent != work_dir:
        opts._directory = work_dir / opts._directory
    if not self.keep_paths_relative:
        opts._directory = opts._directory.resolve(strict=False)

    # Track the directories that need to be created at start of workflow
    self._directory_list = [
        work_dir,
        self.unwrap_options._directory,
    ]

WhirlwindOptions

Bases: BaseModel

User-tunable options for the whirlwind (ww) unwrapper.

ww uses an internal rayon thread pool; pool size is shared across concurrent unwraps (UnwrapOptions.n_parallel_jobs). Empirically ww's MCF is ~40% parallel / 60% serial (Amdahl-limited by the primal-dual phase), so each unwrap saturates around 4 threads — beyond that, added cores give negligible single-IG speedup but concurrent unwraps still benefit from a larger shared pool.

Source code in src/dolphin/workflows/config/_unwrap_options.py

class WhirlwindOptions(BaseModel, extra="forbid"):
    """User-tunable options for the whirlwind (ww) unwrapper.

    ww uses an internal rayon thread pool; pool size is shared across
    concurrent unwraps (``UnwrapOptions.n_parallel_jobs``). Empirically
    ww's MCF is ~40% parallel / 60% serial (Amdahl-limited by the
    primal-dual phase), so each unwrap saturates around 4 threads —
    beyond that, added cores give negligible single-IG speedup but
    concurrent unwraps still benefit from a larger shared pool.
    """

    num_threads: int | None = Field(
        default=None,
        description=(
            "Threads for ww's internal rayon pool. ``None`` = let ww"
            " use whatever ``WHIRLWIND_NUM_THREADS`` / ``RAYON_NUM_THREADS``"
            " is set externally (e.g. via SLURM/``taskset``), falling back"
            " to all CPUs. The pool is *shared* across the"
            " ``n_parallel_jobs`` concurrent unwraps, so the effective"
            " per-IG share is ``num_threads / n_parallel_jobs``."
        ),
        ge=1,
    )

    # --- Spiral persistent-scatterer interpolation pre-pass ------------------
    # Fills low-coherence valid pixels from a Gaussian distance-weighted average
    # of nearby high-coherence phasors before the solve. Like Goldstein, it only
    # informs the MCF: the integer cycle field is transferred back to the
    # original wrapped phase, so per-pixel values are preserved.
    interpolate: bool = Field(
        default=False,
        description=(
            "Enable the spiral PS interpolation pre-pass: every valid pixel"
            " with coherence below ``interp_cutoff`` is filled from nearby"
            " high-coherence pixels before unwrapping."
        ),
    )
    interp_cutoff: float = Field(
        default=0.5,
        description="Coherence below which a valid pixel is interpolated.",
        ge=0.0,
        le=1.0,
    )
    interp_num_neighbors: int = Field(
        default=20,
        description="Nearest high-coherence pixels averaged per interpolated pixel.",
        ge=1,
    )
    interp_max_radius: int = Field(
        default=51,
        description="Maximum search radius (pixels) for the spiral neighbor search.",
        ge=1,
    )
    interp_min_radius: int = Field(
        default=0,
        description="Minimum search radius (pixels); closer neighbors are skipped.",
        ge=0,
    )
    interp_alpha: float = Field(
        default=0.75,
        description="Gaussian distance-weighting falloff for the neighbor average.",
        gt=0.0,
    )

    # --- Connected-component cost / quality knobs ----------------------------
    # An edge becomes a component boundary when its statistical cost is
    # <= cost_threshold. Prefer the physical knobs (sigma / cycle_prob / coh
    # floor) over tuning cost_threshold directly; if more than one is set,
    # whirlwind resolves precedence as sigma > cycle_prob > cost_threshold.
    cost_threshold: int = Field(
        default=50,
        description=(
            "Connected-component boundary threshold in raw cost units. Larger"
            " makes more boundaries and smaller, safer components."
        ),
        ge=0,
    )
    conncomp_sigma: float | None = Field(
        default=None,
        description=(
            "Set ``cost_threshold`` from a Gaussian-equivalent noise level"
            " (~3.5 reproduces the default 50). Higher is stricter (more"
            " boundaries). Takes precedence over ``cost_threshold`` and"
            " ``conncomp_cycle_prob``."
        ),
        gt=0.0,
    )
    conncomp_cycle_prob: float | None = Field(
        default=None,
        description=(
            "Set ``cost_threshold`` from a target per-edge one-cycle-correction"
            " probability (~2.4e-4 matches the default). Lower is stricter."
            " Overridden by ``conncomp_sigma`` if both are set."
        ),
        gt=0.0,
        lt=1.0,
    )
    min_size_px: int = Field(
        default=100,
        description="Discard connected components smaller than this many pixels.",
        ge=1,
    )
    max_ncomps: int = Field(
        default=1024,
        description="Maximum number of connected components to keep (largest first).",
        ge=1,
    )

WorkerSettings

Bases: BaseModel

Settings for controlling CPU/GPU settings and parallelism.

Source code in src/dolphin/workflows/config/_common.py

class WorkerSettings(BaseModel, extra="forbid"):
    """Settings for controlling CPU/GPU settings and parallelism."""

    gpu_enabled: bool = Field(
        False,
        description="Whether to use GPU for processing (if available)",
    )
    threads_per_worker: int = Field(
        1,
        ge=1,
        description=(
            "Number of threads to use per worker. This sets the OMP_NUM_THREADS"
            " environment variable in each python process."
        ),
    )
    n_parallel_bursts: Annotated[
        int,
        tyro.conf.arg(aliases=("--n-parallel-bursts",)),
    ] = Field(
        default=1,
        ge=1,
        description=(
            "If processing separate spatial bursts, number of bursts to run in parallel"
            " for wrapped-phase-estimation. For large, single-swath SLC stacks (e.g."
            " UAVSAR, NISAR), this sets the number of chunks processed in parallel"
            " during phase linking."
        ),
    )
    block_shape: tuple[int, int] = Field(
        (512, 512),
        description="Size (rows, columns) of blocks of data to load at a time.",
    )

YamlModel

Bases: BaseModel

Pydantic model that can be exported to yaml.

Source code in src/dolphin/workflows/config/_yaml_model.py

class YamlModel(BaseModel):
    """Pydantic model that can be exported to yaml."""

    def to_yaml(
        self,
        output_path: Union[Filename, TextIO],
        with_comments: bool = True,
        by_alias: bool = True,
        indent_per_level: int = 2,
    ):
        """Save configuration as a yaml file.

        Used to record the default-filled version of a supplied yaml.

        Parameters
        ----------
        output_path : Pathlike
            Path to the yaml file to save.
        with_comments : bool, default = False.
            Whether to add comments containing the type/descriptions to all fields.
        by_alias : bool, default = False.
            Whether to use the alias names for the fields.
            Passed to pydantic's ``to_json`` method.
            https://docs.pydantic.dev/usage/exporting_models/#modeljson
        indent_per_level : int, default = 2
            Number of spaces to indent per level.

        """
        yaml_obj = self._to_yaml_obj(by_alias=by_alias)

        if with_comments:
            _add_comments(
                yaml_obj,
                self.model_json_schema(by_alias=by_alias),
                indent_per_level=indent_per_level,
            )

        y = YAML()
        # https://yaml.readthedocs.io/en/latest/detail.html#indentation-of-block-sequences
        y.indent(
            offset=indent_per_level,
            mapping=indent_per_level,
            # It is best to always have sequence >= offset + 2 but this is not enforced
            # not following this advice might lead to invalid output.
            sequence=indent_per_level + 2,
        )
        if hasattr(output_path, "write"):
            y.dump(yaml_obj, output_path)
        else:
            with open(output_path, "w") as f:
                y.dump(yaml_obj, f)

    @classmethod
    def from_yaml(cls, yaml_path: Filename):
        """Load a configuration from a yaml file.

        Parameters
        ----------
        yaml_path : Pathlike
            Path to the yaml file to load.

        Returns
        -------
        Config
            Workflow configuration

        """
        y = YAML(typ="safe")
        with open(yaml_path) as f:
            data = y.load(f)

        return cls(**data)

    @classmethod
    def print_yaml_schema(
        cls,
        output_path: Union[Filename, TextIO] = sys.stdout,
        indent_per_level: int = 2,
    ):
        """Print/save an empty configuration with defaults filled in.

        Ignores the required `cslc_file_list` input, so a user can
        inspect all fields.

        Parameters
        ----------
        output_path : Pathlike
            Path or stream to save to the yaml file to.
            By default, prints to stdout.
        indent_per_level : int, default = 2
            Number of spaces to indent per level.

        """
        cls.model_construct().to_yaml(
            output_path, with_comments=True, indent_per_level=indent_per_level
        )

    def _to_yaml_obj(self, by_alias: bool = True) -> CommentedMap:
        # Make the YAML object to add comments to
        # We can't just do `dumps` for some reason, need a stream
        y = YAML()
        ss = StringIO()
        y.dump(json.loads(self.model_dump_json(by_alias=by_alias)), ss)
        return y.load(ss.getvalue())

from_yaml(yaml_path) classmethod

Load a configuration from a yaml file.

Parameters:

Name	Type	Description	Default
yaml_path	Pathlike	Path to the yaml file to load.	required

Returns:

Type	Description
Config	Workflow configuration

Source code in src/dolphin/workflows/config/_yaml_model.py

@classmethod
def from_yaml(cls, yaml_path: Filename):
    """Load a configuration from a yaml file.

    Parameters
    ----------
    yaml_path : Pathlike
        Path to the yaml file to load.

    Returns
    -------
    Config
        Workflow configuration

    """
    y = YAML(typ="safe")
    with open(yaml_path) as f:
        data = y.load(f)

    return cls(**data)

print_yaml_schema(output_path=sys.stdout, indent_per_level=2) classmethod

Print/save an empty configuration with defaults filled in.

Ignores the required cslc_file_list input, so a user can inspect all fields.

Parameters:

Name	Type	Description	Default
output_path	Pathlike	Path or stream to save to the yaml file to. By default, prints to stdout.	stdout
indent_per_level	int	Number of spaces to indent per level.	= 2

Source code in src/dolphin/workflows/config/_yaml_model.py

@classmethod
def print_yaml_schema(
    cls,
    output_path: Union[Filename, TextIO] = sys.stdout,
    indent_per_level: int = 2,
):
    """Print/save an empty configuration with defaults filled in.

    Ignores the required `cslc_file_list` input, so a user can
    inspect all fields.

    Parameters
    ----------
    output_path : Pathlike
        Path or stream to save to the yaml file to.
        By default, prints to stdout.
    indent_per_level : int, default = 2
        Number of spaces to indent per level.

    """
    cls.model_construct().to_yaml(
        output_path, with_comments=True, indent_per_level=indent_per_level
    )

to_yaml(output_path, with_comments=True, by_alias=True, indent_per_level=2)

Save configuration as a yaml file.

Used to record the default-filled version of a supplied yaml.

Parameters:

Name	Type	Description	Default
output_path	Pathlike	Path to the yaml file to save.	required
with_comments	bool	Whether to add comments containing the type/descriptions to all fields.	= False.
by_alias	bool	Whether to use the alias names for the fields. Passed to pydantic's to_json method. https://docs.pydantic.dev/usage/exporting_models/#modeljson	= False.
indent_per_level	int	Number of spaces to indent per level.	= 2

Source code in src/dolphin/workflows/config/_yaml_model.py

def to_yaml(
    self,
    output_path: Union[Filename, TextIO],
    with_comments: bool = True,
    by_alias: bool = True,
    indent_per_level: int = 2,
):
    """Save configuration as a yaml file.

    Used to record the default-filled version of a supplied yaml.

    Parameters
    ----------
    output_path : Pathlike
        Path to the yaml file to save.
    with_comments : bool, default = False.
        Whether to add comments containing the type/descriptions to all fields.
    by_alias : bool, default = False.
        Whether to use the alias names for the fields.
        Passed to pydantic's ``to_json`` method.
        https://docs.pydantic.dev/usage/exporting_models/#modeljson
    indent_per_level : int, default = 2
        Number of spaces to indent per level.

    """
    yaml_obj = self._to_yaml_obj(by_alias=by_alias)

    if with_comments:
        _add_comments(
            yaml_obj,
            self.model_json_schema(by_alias=by_alias),
            indent_per_level=indent_per_level,
        )

    y = YAML()
    # https://yaml.readthedocs.io/en/latest/detail.html#indentation-of-block-sequences
    y.indent(
        offset=indent_per_level,
        mapping=indent_per_level,
        # It is best to always have sequence >= offset + 2 but this is not enforced
        # not following this advice might lead to invalid output.
        sequence=indent_per_level + 2,
    )
    if hasattr(output_path, "write"):
        y.dump(yaml_obj, output_path)
    else:
        with open(output_path, "w") as f:
            y.dump(yaml_obj, f)