aviris module¶

This module contains functions to read and process NASA AVIRIS hyperspectral data. More info about the data can be found at https://aviris.jpl.nasa.gov. A portion of the source code is adapted from the jjmcnelis/aviris-ng-notebooks repository available at https://bit.ly/4bRCgqs. It is licensed under the MIT License. Credit goes to the original author Jack McNelis.

SPDX-FileCopyrightText = [ "2024 Jack McNelis jjmcne@gmail.com", ] SPDX-License-Identifier = "MIT"

`aviris_to_image(dataset, wavelengths=None, method='nearest', output=None, **kwargs)` ¶

Converts an AVIRIS dataset to an image.

Parameters:

Name	Type	Description	Default
`dataset`	`Union[xr.Dataset, str]`	The dataset containing the AVIRIS data or the file path to the dataset.	required
`wavelengths`	`np.ndarray`	The specific wavelengths to select. If None, all wavelengths are selected. Defaults to None.	`None`
`method`	`str`	The method to use for data interpolation. Defaults to "nearest".	`'nearest'`
`output`	`str`	The file path where the image will be saved. If None, the image will be returned as a PIL Image object. Defaults to None.	`None`
`**kwargs`	`Any`	Additional keyword arguments to be passed to `leafmap.array_to_image`.	`{}`

Returns:

Type	Description
`Optional[rasterio.Dataset]`	The image converted from the dataset. If `output` is provided, the image will be saved to the specified file and the function will return None.

Source code in hypercoast/aviris.py

def aviris_to_image(
    dataset: Union[xr.Dataset, str],
    wavelengths: Optional[np.ndarray] = None,
    method: str = "nearest",
    output: Optional[str] = None,
    **kwargs: Any,
):
    """
    Converts an AVIRIS dataset to an image.

    Args:
        dataset (Union[xr.Dataset, str]): The dataset containing the AVIRIS data
            or the file path to the dataset.
        wavelengths (np.ndarray, optional): The specific wavelengths to select. If None, all
            wavelengths are selected. Defaults to None.
        method (str, optional): The method to use for data interpolation.
            Defaults to "nearest".
        output (str, optional): The file path where the image will be saved. If
            None, the image will be returned as a PIL Image object. Defaults to None.
        **kwargs (Any): Additional keyword arguments to be passed to
            `leafmap.array_to_image`.

    Returns:
        Optional[rasterio.Dataset]: The image converted from the dataset. If
            `output` is provided, the image will be saved to the specified file
            and the function will return None.
    """
    from leafmap import array_to_image

    if isinstance(dataset, str):
        dataset = read_aviris(dataset, method=method)

    if wavelengths is not None:
        dataset = dataset.sel(wavelength=wavelengths, method=method)

    return array_to_image(
        dataset["reflectance"],
        output=output,
        transpose=False,
        dtype=np.float32,
        **kwargs,
    )

`extract_aviris(dataset, lat, lon, offset=2.0, allow_remote=False)` ¶

Extracts AVIRIS data from a given xarray Dataset.

Parameters:

Name	Type	Description	Default
`dataset`	`xarray.Dataset`	The dataset containing the AVIRIS data.	required
`lat`	`float`	The latitude of the point to extract.	required
`lon`	`float`	The longitude of the point to extract.	required
`offset`	`float`	The offset from the point to extract. Defaults to 2.0.	`2.0`
`allow_remote`	`bool`	Allow extraction from a remote NetCDF source. Defaults to False because point reads from AVIRIS NetCDF cubes can be very slow over HTTPS/S3.	`False`

Returns:

Type	Description
`xarray.DataArray`	The extracted data.

Source code in hypercoast/aviris.py

def extract_aviris(
    dataset: xr.Dataset,
    lat: float,
    lon: float,
    offset: float = 2.0,
    allow_remote: bool = False,
) -> xr.DataArray:
    """
    Extracts AVIRIS data from a given xarray Dataset.

    Args:
        dataset (xarray.Dataset): The dataset containing the AVIRIS data.
        lat (float): The latitude of the point to extract.
        lon (float): The longitude of the point to extract.
        offset (float, optional): The offset from the point to extract. Defaults to 2.0.
        allow_remote (bool, optional): Allow extraction from a remote NetCDF
            source. Defaults to False because point reads from AVIRIS NetCDF
            cubes can be very slow over HTTPS/S3.

    Returns:
        xarray.DataArray: The extracted data.
    """

    if dataset.attrs.get("_source_is_remote") and not allow_remote:
        source = dataset.attrs.get("_source", "remote AVIRIS NetCDF")
        raise RuntimeError(
            "Interactive spectral extraction from a remote AVIRIS NetCDF can be "
            "very slow because the reflectance cube is chunked in large spatial "
            f"blocks. Download the reflectance NetCDF locally and reopen it with "
            f"read_aviris() before using the spectral widget. Source: {source}"
        )

    crs = _dataset_crs(dataset)

    x, y = convert_coords([[lat, lon]], "epsg:4326", crs)[0]

    da = dataset["reflectance"]

    if "xc" in da.coords and "yc" in da.coords:
        x_con = (da["xc"] > x - offset) & (da["xc"] < x + offset)
        y_con = (da["yc"] > y - offset) & (da["yc"] < y + offset)

        try:
            data = da.where(x_con & y_con, drop=True)
            data = data.mean(dim=["x", "y"])
        except ValueError:
            data = np.nan * np.ones(da.sizes["wavelength"])
    else:
        try:
            x_con = (da["x"] > x - offset) & (da["x"] < x + offset)
            y_con = (da["y"] > y - offset) & (da["y"] < y + offset)
            data = da.where(x_con & y_con, drop=True)
            if data.sizes["x"] == 0 or data.sizes["y"] == 0:
                data = da.sel(x=x, y=y, method="nearest")
            else:
                data = data.mean(dim=["x", "y"])
        except (KeyError, ValueError):
            data = np.nan * np.ones(da.sizes["wavelength"])

    da = xr.DataArray(
        data, dims=["wavelength"], coords={"wavelength": dataset.coords["wavelength"]}
    )

    return da

`get_aviris_asset_url(granule, asset='RFL_ORT', prefer_s3=False)` ¶

Get an AVIRIS asset URL from a CMR/earthaccess granule.

Parameters:

Name	Type	Description	Default
`granule`	`Any`	A CMR granule dictionary, an earthaccess granule, or a STAC-like item dictionary.	required
`asset`	`str`	Asset alias or filename fragment. Common aliases include `"RFL_ORT"`, `"UNC_ORT"`, `"RFL_ORT_QL"`, and `"BROWSE"`. Defaults to `"RFL_ORT"`.	`'RFL_ORT'`
`prefer_s3`	`bool`	Prefer `s3://` links when available. Defaults to False.	`False`

Returns:

Type	Description
`str`	The matching asset URL.

Source code in hypercoast/aviris.py

def get_aviris_asset_url(
    granule: Any,
    asset: str = "RFL_ORT",
    prefer_s3: bool = False,
) -> str:
    """Get an AVIRIS asset URL from a CMR/earthaccess granule.

    Args:
        granule (Any): A CMR granule dictionary, an earthaccess granule, or a
            STAC-like item dictionary.
        asset (str, optional): Asset alias or filename fragment. Common aliases
            include ``"RFL_ORT"``, ``"UNC_ORT"``, ``"RFL_ORT_QL"``, and
            ``"BROWSE"``. Defaults to ``"RFL_ORT"``.
        prefer_s3 (bool, optional): Prefer ``s3://`` links when available.
            Defaults to False.

    Returns:
        str: The matching asset URL.
    """
    asset_key = asset.lower()
    fragments = AVIRIS_ASSET_ALIASES.get(asset_key, (asset,))
    links = _granule_links(granule)
    if not links:
        raise ValueError("No links found in the AVIRIS granule.")

    def score(link: dict) -> int:
        href = str(link.get("href", ""))
        is_s3 = href.startswith("s3://")
        if prefer_s3 and is_s3:
            return 0
        if not prefer_s3 and not is_s3:
            return 0
        return 1

    for link in sorted(links, key=score):
        href = str(link.get("href", ""))
        title = str(link.get("title", ""))
        haystack = f"{href} {title}".lower()
        if any(fragment.lower() in haystack for fragment in fragments):
            return href

    raise ValueError(f"Could not find AVIRIS asset matching {asset!r}.")

`get_aviris_bounds(granule)` ¶

Get AVIRIS granule bounds as [min_lon, min_lat, max_lon, max_lat].

Source code in hypercoast/aviris.py

def get_aviris_bounds(granule: Any) -> Optional[List[float]]:
    """Get AVIRIS granule bounds as [min_lon, min_lat, max_lon, max_lat]."""
    if not isinstance(granule, dict):
        return None

    spatial = None
    umm = granule.get("umm")
    if isinstance(umm, dict):
        spatial = umm.get("SpatialExtent")
    if spatial is None:
        spatial = granule.get("SpatialExtent")

    try:
        polygons = spatial["HorizontalSpatialDomain"]["Geometry"]["GPolygons"]
    except (TypeError, KeyError):
        return None

    lons = []
    lats = []
    for polygon in polygons:
        points = polygon.get("Boundary", {}).get("Points", [])
        for point in points:
            if "Longitude" in point and "Latitude" in point:
                lons.append(float(point["Longitude"]))
                lats.append(float(point["Latitude"]))

    if not lons or not lats:
        return None

    return [min(lons), min(lats), max(lons), max(lats)]

`get_aviris_collection_concept_id(granule)` ¶

Get the CMR collection concept ID from an AVIRIS granule-like object.

Source code in hypercoast/aviris.py

def get_aviris_collection_concept_id(granule: Any) -> Optional[str]:
    """Get the CMR collection concept ID from an AVIRIS granule-like object."""
    if isinstance(granule, dict):
        meta = granule.get("meta")
        if isinstance(meta, dict) and meta.get("collection-concept-id"):
            return meta["collection-concept-id"]
        for key in ("collection-concept-id", "collection_concept_id"):
            if granule.get(key):
                return granule[key]
    return None

`get_aviris_granule_ur(granule)` ¶

Get the CMR GranuleUR from an AVIRIS granule-like object.

Source code in hypercoast/aviris.py

def get_aviris_granule_ur(granule: Any) -> Optional[str]:
    """Get the CMR GranuleUR from an AVIRIS granule-like object."""
    if isinstance(granule, dict):
        umm = granule.get("umm")
        if isinstance(umm, dict) and umm.get("GranuleUR"):
            return umm["GranuleUR"]
        for key in ("GranuleUR", "producer_granule_id", "title"):
            if granule.get(key):
                return granule[key]
    return None

`read_aviris(filepath, wavelengths=None, method='nearest', chunks='auto', engine=None, variable='reflectance', storage_options=None, group='reflectance', **kwargs)` ¶

Reads NASA AVIRIS hyperspectral data and returns an xarray dataset.

Parameters:

Name	Type	Description	Default
`filepath`	`str`	The path or URL to the AVIRIS data.	required
`wavelengths`	`List[float]`	The wavelengths to select. If None, all wavelengths are selected. Defaults to None.	`None`
`method`	`str`	The method to use for selection. Defaults to "nearest".	`'nearest'`
`chunks`	`Any`	Chunking passed to `xarray.open_dataset` for NetCDF inputs. Defaults to `"auto"`.	`'auto'`
`engine`	`str`	Xarray backend engine for NetCDF inputs. Remote NetCDF sources default to `"h5netcdf"`.	`None`
`variable`	`str`	Data variable to read from NetCDF inputs. Defaults to `"reflectance"`.	`'reflectance'`
`storage_options`	`dict`	Options passed to `fsspec.open` for remote NetCDF inputs, such as S3 or HTTP credentials.	`None`
`group`	`str`	NetCDF group containing the reflectance variable. Defaults to `"reflectance"` for AVIRIS-3/5 products. If the group is not present, the root dataset is opened.	`'reflectance'`
`**kwargs`	`Any`	Additional arguments to pass to the selection method.	`{}`

Returns:

Type	Description
`xr.Dataset`	The dataset containing the reflectance data.

Source code in hypercoast/aviris.py

def read_aviris(
    filepath: str,
    wavelengths: Optional[List[float]] = None,
    method: str = "nearest",
    chunks: Any = "auto",
    engine: Optional[str] = None,
    variable: str = "reflectance",
    storage_options: Optional[dict] = None,
    group: Optional[str] = "reflectance",
    **kwargs: Any,
) -> xr.Dataset:
    """
    Reads NASA AVIRIS hyperspectral data and returns an xarray dataset.

    Args:
        filepath (str): The path or URL to the AVIRIS data.
        wavelengths (List[float], optional): The wavelengths to select. If None,
            all wavelengths are selected. Defaults to None.
        method (str, optional): The method to use for selection. Defaults to
            "nearest".
        chunks (Any, optional): Chunking passed to ``xarray.open_dataset`` for
            NetCDF inputs. Defaults to ``"auto"``.
        engine (str, optional): Xarray backend engine for NetCDF inputs. Remote
            NetCDF sources default to ``"h5netcdf"``.
        variable (str, optional): Data variable to read from NetCDF inputs.
            Defaults to ``"reflectance"``.
        storage_options (dict, optional): Options passed to ``fsspec.open`` for
            remote NetCDF inputs, such as S3 or HTTP credentials.
        group (str, optional): NetCDF group containing the reflectance variable.
            Defaults to ``"reflectance"`` for AVIRIS-3/5 products. If the group
            is not present, the root dataset is opened.
        **kwargs (Any): Additional arguments to pass to the selection method.

    Returns:
        xr.Dataset: The dataset containing the reflectance data.
    """
    if not isinstance(filepath, (str, os.PathLike)) and not hasattr(filepath, "read"):
        filepath = get_aviris_asset_url(filepath, asset="RFL_ORT")

    if isinstance(filepath, os.PathLike):
        filepath = os.fspath(filepath)

    file_name = _source_name(filepath)
    if _is_netcdf_path(filepath) or _is_netcdf_path(file_name):
        ds = _open_aviris_netcdf(
            filepath,
            chunks=chunks,
            engine=engine,
            storage_options=storage_options,
            group=group,
        )
        ds = _normalize_aviris_netcdf(
            ds, wavelengths=wavelengths, method=method, variable=variable, **kwargs
        )
        if file_name:
            ds.attrs["_source"] = file_name
            ds.attrs["_source_is_remote"] = _is_remote_path(file_name)
        return ds

    if filepath.endswith(".hdr"):
        filepath = filepath.replace(".hdr", "")

    ds = xr.open_dataset(filepath, engine="rasterio")

    wavelength = ds["wavelength"].values.tolist()
    wavelength = [round(num, 2) for num in wavelength]

    cols = ds.x.size
    rows = ds.y.size

    rio_transform = ds.rio.transform()
    geo_transform = list(rio_transform)[:6]

    # get the raster geotransform as its component parts
    xres, _, xmin, _, yres, ymax = geo_transform

    # generate coordinate arrays
    xarr = np.array([xmin + i * xres for i in range(0, cols)])
    yarr = np.array([ymax + i * yres for i in range(0, rows)])

    ds["y"] = xr.DataArray(
        data=yarr,
        dims=("y"),
        name="y",
        attrs=dict(
            units="m",
            standard_name="projection_y_coordinate",
            long_name="y coordinate of projection",
        ),
    )

    ds["x"] = xr.DataArray(
        data=xarr,
        dims=("x"),
        name="x",
        attrs=dict(
            units="m",
            standard_name="projection_x_coordinate",
            long_name="x coordinate of projection",
        ),
    )

    global_atts = ds.attrs
    global_atts["Conventions"] = "CF-1.6"
    ds.attrs = dict(
        units="unitless",
        _FillValue=-9999,
        grid_mapping="crs",
        standard_name="reflectance",
        long_name="atmospherically corrected surface reflectance",
    )
    ds.attrs.update(global_atts)

    ds = ds.transpose("y", "x", "band")
    ds = ds.drop_vars(["wavelength"])
    ds = ds.rename({"band": "wavelength", "band_data": "reflectance"})
    ds.coords["wavelength"] = wavelength
    ds.attrs["crs"] = ds.rio.crs.to_string()
    ds.rio.write_transform(rio_transform)

    if wavelengths is not None:
        ds = ds.sel(wavelength=wavelengths, method=method, **kwargs)
    return ds

aviris module¶

aviris_to_image(dataset, wavelengths=None, method='nearest', output=None, **kwargs) ¶

extract_aviris(dataset, lat, lon, offset=2.0, allow_remote=False) ¶

get_aviris_asset_url(granule, asset='RFL_ORT', prefer_s3=False) ¶

get_aviris_bounds(granule) ¶

get_aviris_collection_concept_id(granule) ¶

get_aviris_granule_ur(granule) ¶

read_aviris(filepath, wavelengths=None, method='nearest', chunks='auto', engine=None, variable='reflectance', storage_options=None, group='reflectance', **kwargs) ¶

`aviris_to_image(dataset, wavelengths=None, method='nearest', output=None, **kwargs)` ¶

`extract_aviris(dataset, lat, lon, offset=2.0, allow_remote=False)` ¶

`get_aviris_asset_url(granule, asset='RFL_ORT', prefer_s3=False)` ¶

`get_aviris_bounds(granule)` ¶

`get_aviris_collection_concept_id(granule)` ¶

`get_aviris_granule_ur(granule)` ¶

`read_aviris(filepath, wavelengths=None, method='nearest', chunks='auto', engine=None, variable='reflectance', storage_options=None, group='reflectance', **kwargs)` ¶