Source code for pystac.extensions.eo

"""Implements the :stac-ext:`Electro-Optical Extension <eo>`."""

from __future__ import annotations

import warnings
from collections.abc import Iterable
from typing import (
    Any,
    Generic,
    Literal,
    TypeVar,
    Union,
    cast,
)

import pystac
from pystac.extensions import item_assets, projection, view
from pystac.extensions.base import (
    ExtensionManagementMixin,
    PropertiesExtension,
    SummariesExtension,
)
from pystac.extensions.hooks import ExtensionHooks
from pystac.serialization.identify import STACJSONDescription, STACVersionID
from pystac.summaries import RangeSummary
from pystac.utils import get_required, map_opt

T = TypeVar("T", pystac.Item, pystac.Asset, item_assets.AssetDefinition)

SCHEMA_URI: str = "https://stac-extensions.github.io/eo/v1.1.0/schema.json"
SCHEMA_URIS: list[str] = [
    "https://stac-extensions.github.io/eo/v1.0.0/schema.json",
    SCHEMA_URI,
]
PREFIX: str = "eo:"

# Field names
BANDS_PROP: str = PREFIX + "bands"
CLOUD_COVER_PROP: str = PREFIX + "cloud_cover"
SNOW_COVER_PROP: str = PREFIX + "snow_cover"


[docs]def validated_percentage(v: float | None) -> float | None: if v is not None and not isinstance(v, (float, int)) or isinstance(v, bool): raise ValueError(f"Invalid percentage: {v} must be number") if v is not None and not 0 <= v <= 100: raise ValueError(f"Invalid percentage: {v} must be between 0 and 100") return v
[docs]class Band: """Represents Band information attached to an Item that implements the eo extension. Use :meth:`Band.create` to create a new Band. """ properties: dict[str, Any] def __init__(self, properties: dict[str, Any]) -> None: self.properties = properties
[docs] def apply( self, name: str, common_name: str | None = None, description: str | None = None, center_wavelength: float | None = None, full_width_half_max: float | None = None, solar_illumination: float | None = None, ) -> None: """ Sets the properties for this Band. Args: name : The name of the band (e.g., "B01", "B02", "B1", "B5", "QA"). common_name : The name commonly used to refer to the band to make it easier to search for bands across instruments. See the :stac-ext:`list of accepted common names <eo#common-band-names>`. description : Description to fully explain the band. center_wavelength : The center wavelength of the band, in micrometers (μm). full_width_half_max : Full width at half maximum (FWHM). The width of the band, as measured at half the maximum transmission, in micrometers (μm). solar_illumination: The solar illumination of the band, as measured at half the maximum transmission, in W/m2/micrometers. """ self.name = name self.common_name = common_name self.description = description self.center_wavelength = center_wavelength self.full_width_half_max = full_width_half_max self.solar_illumination = solar_illumination
[docs] @classmethod def create( cls, name: str, common_name: str | None = None, description: str | None = None, center_wavelength: float | None = None, full_width_half_max: float | None = None, solar_illumination: float | None = None, ) -> Band: """ Creates a new band. Args: name : The name of the band (e.g., "B01", "B02", "B1", "B5", "QA"). common_name : The name commonly used to refer to the band to make it easier to search for bands across instruments. See the :stac-ext:`list of accepted common names <eo#common-band-names>`. description : Description to fully explain the band. center_wavelength : The center wavelength of the band, in micrometers (μm). full_width_half_max : Full width at half maximum (FWHM). The width of the band, as measured at half the maximum transmission, in micrometers (μm). solar_illumination: The solar illumination of the band, as measured at half the maximum transmission, in W/m2/micrometers. """ b = cls({}) b.apply( name=name, common_name=common_name, description=description, center_wavelength=center_wavelength, full_width_half_max=full_width_half_max, solar_illumination=solar_illumination, ) return b
@property def name(self) -> str: """Get or sets the name of the band (e.g., "B01", "B02", "B1", "B5", "QA"). Returns: str """ return get_required(self.properties.get("name"), self, "name") @name.setter def name(self, v: str) -> None: self.properties["name"] = v @property def common_name(self) -> str | None: """Get or sets the name commonly used to refer to the band to make it easier to search for bands across instruments. See the :stac-ext:`list of accepted common names <eo#common-band-names>`. Returns: Optional[str] """ return self.properties.get("common_name") @common_name.setter def common_name(self, v: str | None) -> None: if v is not None: self.properties["common_name"] = v else: self.properties.pop("common_name", None) @property def description(self) -> str | None: """Get or sets the description to fully explain the band. CommonMark 0.29 syntax MAY be used for rich text representation. Returns: str """ return self.properties.get("description") @description.setter def description(self, v: str | None) -> None: if v is not None: self.properties["description"] = v else: self.properties.pop("description", None) @property def center_wavelength(self) -> float | None: """Get or sets the center wavelength of the band, in micrometers (μm). Returns: float """ return self.properties.get("center_wavelength") @center_wavelength.setter def center_wavelength(self, v: float | None) -> None: if v is not None: self.properties["center_wavelength"] = v else: self.properties.pop("center_wavelength", None) @property def full_width_half_max(self) -> float | None: """Get or sets the full width at half maximum (FWHM). The width of the band, as measured at half the maximum transmission, in micrometers (μm). Returns: [float] """ return self.properties.get("full_width_half_max") @full_width_half_max.setter def full_width_half_max(self, v: float | None) -> None: if v is not None: self.properties["full_width_half_max"] = v else: self.properties.pop("full_width_half_max", None) @property def solar_illumination(self) -> float | None: """Get or sets the solar illumination of the band, as measured at half the maximum transmission, in W/m2/micrometers. Returns: [float] """ return self.properties.get("solar_illumination") @solar_illumination.setter def solar_illumination(self, v: float | None) -> None: if v is not None: self.properties["solar_illumination"] = v else: self.properties.pop("solar_illumination", None) def __repr__(self) -> str: return f"<Band name={self.name}>"
[docs] def to_dict(self) -> dict[str, Any]: """Returns this band as a dictionary. Returns: dict: The serialization of this Band. """ return self.properties
[docs] @staticmethod def band_range(common_name: str) -> tuple[float, float] | None: """Gets the band range for a common band name. Args: common_name : The common band name. Must be one of the :stac-ext:`list of accepted common names <eo#common-band-names>`. Returns: Tuple[float, float] or None: The band range for this name as (min, max), or None if this is not a recognized common name. """ name_to_range = { "coastal": (0.40, 0.45), "blue": (0.45, 0.50), "green": (0.50, 0.60), "red": (0.60, 0.70), "yellow": (0.58, 0.62), "pan": (0.50, 0.70), "rededge": (0.70, 0.75), "nir": (0.75, 1.00), "nir08": (0.75, 0.90), "nir09": (0.85, 1.05), "cirrus": (1.35, 1.40), "swir16": (1.55, 1.75), "swir22": (2.10, 2.30), "lwir": (10.5, 12.5), "lwir11": (10.5, 11.5), "lwir12": (11.5, 12.5), } return name_to_range.get(common_name)
[docs] @staticmethod def band_description(common_name: str) -> str | None: """Returns a description of the band for one with a common name. Args: common_name : The common band name. Must be one of the :stac-ext:`list of accepted common names <eo#common-band-names>`. Returns: str or None: If a recognized common name, returns a description including the band range. Otherwise returns None. """ r = Band.band_range(common_name) if r is not None: return f"Common name: {common_name}, Range: {r[0]} to {r[1]}" return None
[docs]class EOExtension( Generic[T], PropertiesExtension, ExtensionManagementMixin[Union[pystac.Item, pystac.Collection]], ): """An abstract class that can be used to extend the properties of an :class:`~pystac.Item` or :class:`~pystac.Asset` with properties from the :stac-ext:`Electro-Optical Extension <eo>`. This class is generic over the type of STAC Object to be extended (e.g. :class:`~pystac.Item`, :class:`~pystac.Asset`). To create a concrete instance of :class:`EOExtension`, use the :meth:`EOExtension.ext` method. For example: .. code-block:: python >>> item: pystac.Item = ... >>> eo_ext = EOExtension.ext(item) """ name: Literal["eo"] = "eo"
[docs] def apply( self, bands: list[Band] | None = None, cloud_cover: float | None = None, snow_cover: float | None = None, ) -> None: """Applies Electro-Optical Extension properties to the extended :class:`~pystac.Item` or :class:`~pystac.Asset`. Args: bands : A list of available bands where each item is a :class:`~Band` object. If given, requires at least one band. cloud_cover : The estimate of cloud cover as a percentage (0-100) of the entire scene. If not available the field should not be provided. snow_cover : The estimate of snow cover as a percentage (0-100) of the entire scene. If not available the field should not be provided. """ self.bands = bands self.cloud_cover = validated_percentage(cloud_cover) self.snow_cover = validated_percentage(snow_cover)
@property def bands(self) -> list[Band] | None: """Gets or sets a list of available bands where each item is a :class:`~Band` object (or ``None`` if no bands have been set). If not available the field should not be provided. """ return self._get_bands() @bands.setter def bands(self, v: list[Band] | None) -> None: self._set_property( BANDS_PROP, map_opt(lambda bands: [b.to_dict() for b in bands], v) ) def _get_bands(self) -> list[Band] | None: return map_opt( lambda bands: [Band(b) for b in bands], self._get_property(BANDS_PROP, list[dict[str, Any]]), ) @property def cloud_cover(self) -> float | None: """Get or sets the estimate of cloud cover as a percentage (0-100) of the entire scene. If not available the field should not be provided. Returns: float or None """ return self._get_property(CLOUD_COVER_PROP, float) @cloud_cover.setter def cloud_cover(self, v: float | None) -> None: self._set_property(CLOUD_COVER_PROP, validated_percentage(v), pop_if_none=True) @property def snow_cover(self) -> float | None: """Get or sets the estimate of snow cover as a percentage (0-100) of the entire scene. If not available the field should not be provided. Returns: float or None """ return self._get_property(SNOW_COVER_PROP, float) @snow_cover.setter def snow_cover(self, v: float | None) -> None: self._set_property(SNOW_COVER_PROP, validated_percentage(v), pop_if_none=True)
[docs] @classmethod def get_schema_uri(cls) -> str: return SCHEMA_URI
[docs] @classmethod def get_schema_uris(cls) -> list[str]: warnings.warn( "get_schema_uris is deprecated and will be removed in v2", DeprecationWarning, ) return SCHEMA_URIS
[docs] @classmethod def ext(cls, obj: T, add_if_missing: bool = False) -> EOExtension[T]: """Extends the given STAC Object with properties from the :stac-ext:`Electro-Optical Extension <eo>`. This extension can be applied to instances of :class:`~pystac.Item` or :class:`~pystac.Asset`. Raises: pystac.ExtensionTypeError : If an invalid object type is passed. """ if isinstance(obj, pystac.Item): cls.ensure_has_extension(obj, add_if_missing) return cast(EOExtension[T], ItemEOExtension(obj)) elif isinstance(obj, pystac.Asset): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(EOExtension[T], AssetEOExtension(obj)) elif isinstance(obj, item_assets.AssetDefinition): cls.ensure_owner_has_extension(obj, add_if_missing) return cast(EOExtension[T], ItemAssetsEOExtension(obj)) else: raise pystac.ExtensionTypeError(cls._ext_error_message(obj))
[docs] @classmethod def summaries( cls, obj: pystac.Collection, add_if_missing: bool = False ) -> SummariesEOExtension: """Returns the extended summaries object for the given collection.""" cls.ensure_has_extension(obj, add_if_missing) return SummariesEOExtension(obj)
[docs]class ItemEOExtension(EOExtension[pystac.Item]): """A concrete implementation of :class:`EOExtension` on an :class:`~pystac.Item` that extends the properties of the Item to include properties defined in the :stac-ext:`Electro-Optical Extension <eo>`. This class should generally not be instantiated directly. Instead, call :meth:`EOExtension.ext` on an :class:`~pystac.Item` to extend it. """ item: pystac.Item """The :class:`~pystac.Item` being extended.""" properties: dict[str, Any] """The :class:`~pystac.Item` properties, including extension properties.""" def __init__(self, item: pystac.Item): self.item = item self.properties = item.properties def _get_bands(self) -> list[Band] | None: """Get or sets a list of :class:`~pystac.Band` objects that represent the available bands. """ bands = self._get_property(BANDS_PROP, list[dict[str, Any]]) # get assets with eo:bands even if not in item if bands is None: asset_bands: list[dict[str, Any]] = [] for _, value in self.item.get_assets().items(): if BANDS_PROP in value.extra_fields: asset_bands.extend( cast(list[dict[str, Any]], value.extra_fields.get(BANDS_PROP)) ) if any(asset_bands): bands = asset_bands if bands is not None: return [Band(b) for b in bands] return None
[docs] def get_assets( self, name: str | None = None, common_name: str | None = None, ) -> dict[str, pystac.Asset]: """Get the item's assets where eo:bands are defined. Args: name: If set, filter the assets such that only those with a matching ``eo:band.name`` are returned. common_name: If set, filter the assets such that only those with a matching ``eo:band.common_name`` are returned. Returns: Dict[str, Asset]: A dictionary of assets that match ``name`` and/or ``common_name`` if set or else all of this item's assets were eo:bands are defined. """ kwargs = {"name": name, "common_name": common_name} return { key: asset for key, asset in self.item.get_assets().items() if BANDS_PROP in asset.extra_fields and all( v is None or any(v == b.get(k) for b in asset.extra_fields[BANDS_PROP]) for k, v in kwargs.items() ) }
def __repr__(self) -> str: return f"<ItemEOExtension Item id={self.item.id}>"
[docs]class AssetEOExtension(EOExtension[pystac.Asset]): """A concrete implementation of :class:`EOExtension` on an :class:`~pystac.Asset` that extends the Asset fields to include properties defined in the :stac-ext:`Electro-Optical Extension <eo>`. This class should generally not be instantiated directly. Instead, call :meth:`EOExtension.ext` on an :class:`~pystac.Asset` to extend it. """ asset_href: str """The ``href`` value of the :class:`~pystac.Asset` being extended.""" properties: dict[str, Any] """The :class:`~pystac.Asset` fields, including extension properties.""" additional_read_properties: Iterable[dict[str, Any]] | None = None """If present, this will be a list containing 1 dictionary representing the properties of the owning :class:`~pystac.Item`.""" def _get_bands(self) -> list[Band] | None: if BANDS_PROP not in self.properties: return None return list( map( lambda band: Band(band), cast(list[dict[str, Any]], self.properties.get(BANDS_PROP)), ) ) def __init__(self, asset: pystac.Asset): self.asset_href = asset.href self.properties = asset.extra_fields if asset.owner and isinstance(asset.owner, pystac.Item): self.additional_read_properties = [asset.owner.properties] def __repr__(self) -> str: return f"<AssetEOExtension Asset href={self.asset_href}>"
[docs]class ItemAssetsEOExtension(EOExtension[item_assets.AssetDefinition]): properties: dict[str, Any] asset_defn: item_assets.AssetDefinition def _get_bands(self) -> list[Band] | None: if BANDS_PROP not in self.properties: return None return list( map( lambda band: Band(band), cast(list[dict[str, Any]], self.properties.get(BANDS_PROP)), ) ) def __init__(self, item_asset: item_assets.AssetDefinition): self.asset_defn = item_asset self.properties = item_asset.properties
[docs]class SummariesEOExtension(SummariesExtension): """A concrete implementation of :class:`~SummariesExtension` that extends the ``summaries`` field of a :class:`~pystac.Collection` to include properties defined in the :stac-ext:`Electro-Optical Extension <eo>`. """ @property def bands(self) -> list[Band] | None: """Get or sets the summary of :attr:`EOExtension.bands` values for this Collection. """ return map_opt( lambda bands: [Band(b) for b in bands], self.summaries.get_list(BANDS_PROP), ) @bands.setter def bands(self, v: list[Band] | None) -> None: self._set_summary(BANDS_PROP, map_opt(lambda x: [b.to_dict() for b in x], v)) @property def cloud_cover(self) -> RangeSummary[float] | None: """Get or sets the summary of :attr:`EOExtension.cloud_cover` values for this Collection. """ return self.summaries.get_range(CLOUD_COVER_PROP) @cloud_cover.setter def cloud_cover(self, v: RangeSummary[float] | None) -> None: self._set_summary(CLOUD_COVER_PROP, v) @property def snow_cover(self) -> RangeSummary[float] | None: """Get or sets the summary of :attr:`EOExtension.snow_cover` values for this Collection. """ return self.summaries.get_range(SNOW_COVER_PROP) @snow_cover.setter def snow_cover(self, v: RangeSummary[float] | None) -> None: self._set_summary(SNOW_COVER_PROP, v)
[docs]class EOExtensionHooks(ExtensionHooks): schema_uri: str = SCHEMA_URI prev_extension_ids = { "eo", *[uri for uri in SCHEMA_URIS if uri != SCHEMA_URI], } stac_object_types = {pystac.STACObjectType.ITEM}
[docs] def migrate( self, obj: dict[str, Any], version: STACVersionID, info: STACJSONDescription ) -> None: if version < "0.9": # Some eo fields became common_metadata if ( "eo:platform" in obj["properties"] and "platform" not in obj["properties"] ): obj["properties"]["platform"] = obj["properties"]["eo:platform"] del obj["properties"]["eo:platform"] if ( "eo:instrument" in obj["properties"] and "instruments" not in obj["properties"] ): obj["properties"]["instruments"] = [obj["properties"]["eo:instrument"]] del obj["properties"]["eo:instrument"] if ( "eo:constellation" in obj["properties"] and "constellation" not in obj["properties"] ): obj["properties"]["constellation"] = obj["properties"][ "eo:constellation" ] del obj["properties"]["eo:constellation"] # Some eo fields became view extension fields eo_to_view_fields = [ "off_nadir", "azimuth", "incidence_angle", "sun_azimuth", "sun_elevation", ] for field in eo_to_view_fields: if f"eo:{field}" in obj["properties"]: if "stac_extensions" not in obj: obj["stac_extensions"] = [] if view.SCHEMA_URI not in obj["stac_extensions"]: obj["stac_extensions"].append(view.SCHEMA_URI) if f"view:{field}" not in obj["properties"]: obj["properties"][f"view:{field}"] = obj["properties"][ f"eo:{field}" ] del obj["properties"][f"eo:{field}"] # eo:epsg became proj:epsg eo_epsg = PREFIX + "epsg" proj_epsg = projection.PREFIX + "epsg" if eo_epsg in obj["properties"] and proj_epsg not in obj["properties"]: obj["properties"][proj_epsg] = obj["properties"].pop(eo_epsg) obj["stac_extensions"] = obj.get("stac_extensions", []) if ( projection.ProjectionExtension.get_schema_uri() not in obj["stac_extensions"] ): obj["stac_extensions"].append( projection.ProjectionExtension.get_schema_uri() ) if not any(prop.startswith(PREFIX) for prop in obj["properties"]): obj["stac_extensions"].remove(EOExtension.get_schema_uri()) if version < "1.0.0-beta.1" and info.object_type == pystac.STACObjectType.ITEM: # gsd moved from eo to common metadata if "eo:gsd" in obj["properties"]: obj["properties"]["gsd"] = obj["properties"]["eo:gsd"] del obj["properties"]["eo:gsd"] # The way bands were declared in assets changed. # In 1.0.0-beta.1 they are inlined into assets as # opposed to having indices back into a property-level array. if "eo:bands" in obj["properties"]: bands = obj["properties"]["eo:bands"] for asset in obj["assets"].values(): if "eo:bands" in asset: new_bands: list[dict[str, Any]] = [] for band_index in asset["eo:bands"]: new_bands.append(bands[band_index]) asset["eo:bands"] = new_bands super().migrate(obj, version, info)
EO_EXTENSION_HOOKS: ExtensionHooks = EOExtensionHooks()