workflows module

Coastal workflow presets for hyperspectral datasets.

WorkflowPreset dataclass

Describe a coastal workflow preset.

Parameters:

Name	Type	Description	Default
name	str	Workflow identifier.	required
description	str	Human-readable workflow description.	required
kind	str	Workflow algorithm kind.	required
wavelengths	Dict[str, float]	Named wavelength parameters in nanometers.	required
variable	Optional[str]	Optional default data variable.	None

Source code in hypercoast/workflows.py

@dataclass(frozen=True)
class WorkflowPreset:
    """Describe a coastal workflow preset.

    Args:
        name: Workflow identifier.
        description: Human-readable workflow description.
        kind: Workflow algorithm kind.
        wavelengths: Named wavelength parameters in nanometers.
        variable: Optional default data variable.
    """

    name: str
    description: str
    kind: str
    wavelengths: Dict[str, float]
    variable: Optional[str] = None

    def as_dict(self) -> Dict[str, Any]:
        """Return serializable workflow metadata."""
        return asdict(self)

as_dict(self)

Return serializable workflow metadata.

Source code in hypercoast/workflows.py

def as_dict(self) -> Dict[str, Any]:
    """Return serializable workflow metadata."""
    return asdict(self)

apply_workflow(data, workflow, variable=None, **kwargs)

Apply a named coastal workflow to hyperspectral data.

Parameters:

Name	Type	Description	Default
data	xr.Dataset | xr.DataArray	Input dataset or data array.	required
workflow	str	Workflow preset name.	required
variable	Optional[str]	Optional data variable for xarray datasets.	None
**kwargs	Any	Optional workflow overrides.	{}

Returns:

Type	Description
xr.DataArray	Workflow output.

Source code in hypercoast/workflows.py

def apply_workflow(
    data: xr.Dataset | xr.DataArray,
    workflow: str,
    variable: Optional[str] = None,
    **kwargs: Any,
) -> xr.DataArray:
    """Apply a named coastal workflow to hyperspectral data.

    Args:
        data: Input dataset or data array.
        workflow: Workflow preset name.
        variable: Optional data variable for xarray datasets.
        **kwargs: Optional workflow overrides.

    Returns:
        xr.DataArray: Workflow output.
    """
    key = workflow.lower().strip()
    if key not in WORKFLOW_PRESETS:
        available = ", ".join(sorted(WORKFLOW_PRESETS))
        raise KeyError(
            f"Unknown workflow '{workflow}'. Available workflows: {available}"
        )

    preset = WORKFLOW_PRESETS[key]
    da = _as_data_array(data, variable or preset.variable)
    wavelengths = dict(preset.wavelengths)
    wavelengths.update(kwargs.pop("wavelengths", {}) or {})

    if preset.kind == "normalized_difference":
        return normalized_difference(da, wavelengths["a"], wavelengths["b"])
    if preset.kind == "ratio":
        return band_ratio(da, wavelengths["a"], wavelengths["b"])
    if preset.kind == "spectral_anomaly":
        return spectral_anomaly(da)
    raise ValueError(f"Unsupported workflow kind: {preset.kind}")

band_ratio(data, numerator, denominator, variable=None)

Calculate a ratio between two wavelengths.

Parameters:

Name	Type	Description	Default
data	xr.Dataset | xr.DataArray	Input dataset or data array.	required
numerator	float	Numerator wavelength in nanometers.	required
denominator	float	Denominator wavelength in nanometers.	required
variable	Optional[str]	Optional data variable for xarray datasets.	None

Returns:

Type	Description
xr.DataArray	Ratio output.

Source code in hypercoast/workflows.py

def band_ratio(
    data: xr.Dataset | xr.DataArray,
    numerator: float,
    denominator: float,
    variable: Optional[str] = None,
) -> xr.DataArray:
    """Calculate a ratio between two wavelengths.

    Args:
        data: Input dataset or data array.
        numerator: Numerator wavelength in nanometers.
        denominator: Denominator wavelength in nanometers.
        variable: Optional data variable for xarray datasets.

    Returns:
        xr.DataArray: Ratio output.
    """
    da = _as_data_array(data, variable)
    a = _select_wavelength(da, numerator)
    b = _select_wavelength(da, denominator)
    with np.errstate(divide="ignore", invalid="ignore"):
        result = a / b
    result.name = f"ratio_{int(numerator)}_{int(denominator)}"
    return result

list_workflows()

Return serializable workflow preset metadata.

Returns:

Type	Description
dict	Mapping of workflow names to metadata.

Source code in hypercoast/workflows.py

def list_workflows() -> Dict[str, Dict[str, Any]]:
    """Return serializable workflow preset metadata.

    Returns:
        dict: Mapping of workflow names to metadata.
    """
    return {name: preset.as_dict() for name, preset in WORKFLOW_PRESETS.items()}

normalized_difference(data, wavelength_a, wavelength_b, variable=None)

Calculate a normalized difference between two wavelengths.

Parameters:

Name	Type	Description	Default
data	xr.Dataset | xr.DataArray	Input dataset or data array.	required
wavelength_a	float	First wavelength in nanometers.	required
wavelength_b	float	Second wavelength in nanometers.	required
variable	Optional[str]	Optional data variable for xarray datasets.	None

Returns:

Type	Description
xr.DataArray	Normalized difference output.

Source code in hypercoast/workflows.py

def normalized_difference(
    data: xr.Dataset | xr.DataArray,
    wavelength_a: float,
    wavelength_b: float,
    variable: Optional[str] = None,
) -> xr.DataArray:
    """Calculate a normalized difference between two wavelengths.

    Args:
        data: Input dataset or data array.
        wavelength_a: First wavelength in nanometers.
        wavelength_b: Second wavelength in nanometers.
        variable: Optional data variable for xarray datasets.

    Returns:
        xr.DataArray: Normalized difference output.
    """
    da = _as_data_array(data, variable)
    a = _select_wavelength(da, wavelength_a)
    b = _select_wavelength(da, wavelength_b)
    with np.errstate(divide="ignore", invalid="ignore"):
        result = (a - b) / (a + b)
    result.name = f"nd_{int(wavelength_a)}_{int(wavelength_b)}"
    return result

spectral_anomaly(data, variable=None)

Calculate the maximum absolute spectral z-score per pixel.

Parameters:

Name	Type	Description	Default
data	xr.Dataset | xr.DataArray	Input dataset or data array.	required
variable	Optional[str]	Optional data variable for xarray datasets.	None

Returns:

Type	Description
xr.DataArray	Maximum absolute spectral anomaly score.

Source code in hypercoast/workflows.py

def spectral_anomaly(
    data: xr.Dataset | xr.DataArray,
    variable: Optional[str] = None,
) -> xr.DataArray:
    """Calculate the maximum absolute spectral z-score per pixel.

    Args:
        data: Input dataset or data array.
        variable: Optional data variable for xarray datasets.

    Returns:
        xr.DataArray: Maximum absolute spectral anomaly score.
    """
    da = _as_data_array(data, variable)
    dim = _wavelength_dim(da)
    mean = da.mean(dim=dim, skipna=True)
    std = da.std(dim=dim, skipna=True)
    with np.errstate(divide="ignore", invalid="ignore"):
        zscore = (da - mean) / std
    result = np.abs(zscore).max(dim=dim, skipna=True)
    result.name = "spectral_anomaly"
    return result