class Hls2(_NasaHlsBase):
"""Combined NASA HLS v2.0 time-series datasource with semantic band names."""
# Internal label for the combined datasource; CMR STAC searches use
# SOURCE_TO_COLLECTION below rather than a real "HLS2" collection.
COLLECTION_NAME = "HLS2"
SOURCE_TO_COLLECTION = {
"sentinel": Hls2S30.COLLECTION_NAME,
"landsat": Hls2L30.COLLECTION_NAME,
}
COLLECTION_TO_SOURCE = {
collection_name: source
for source, collection_name in SOURCE_TO_COLLECTION.items()
}
SENTINEL_SEMANTIC_BANDS = {
"coastal": "B01",
"blue": "B02",
"green": "B03",
"red": "B04",
"nir": "B08",
"cirrus": "B10",
"swir16": "B11",
"swir22": "B12",
"fmask": "Fmask",
"solar_azimuth": "SAA",
"solar_zenith": "SZA",
"view_azimuth": "VAA",
"view_zenith": "VZA",
}
LANDSAT_SEMANTIC_BANDS = {
"coastal": "B01",
"blue": "B02",
"green": "B03",
"red": "B04",
"nir": "B05",
"cirrus": "B09",
"swir16": "B06",
"swir22": "B07",
"fmask": "Fmask",
"solar_azimuth": "SAA",
"solar_zenith": "SZA",
"view_azimuth": "VAA",
"view_zenith": "VZA",
}
DEFAULT_BANDS = [
"coastal",
"blue",
"green",
"red",
"nir",
"swir16",
"swir22",
]
@classmethod
def _asset_key_map_for_collection(cls, collection_name: str) -> dict[str, str]:
if collection_name == Hls2S30.COLLECTION_NAME:
return cls.SENTINEL_SEMANTIC_BANDS
if collection_name == Hls2L30.COLLECTION_NAME:
return cls.LANDSAT_SEMANTIC_BANDS
raise ValueError(f"unsupported HLS2 collection {collection_name!r}")
@classmethod
def _normalize_band_name(cls, band: str) -> str:
if band not in cls.SENTINEL_SEMANTIC_BANDS:
raise ValueError(
f"unsupported Hls2 band '{band}'. Use one of "
f"{sorted(cls.SENTINEL_SEMANTIC_BANDS.keys())}."
)
return band
def __init__(
self,
band_names: list[str] | None = None,
sources: list[str] | None = None,
query: dict[str, Any] | None = None,
sort_by: str | None = "datetime",
sort_ascending: bool = True,
timeout: timedelta = timedelta(seconds=30),
earthdata_token: str | None = None,
s3_credentials_url: str = _NasaHlsBase.S3_CREDENTIALS_URL,
context: DataSourceContext = DataSourceContext(),
) -> None:
"""Create a combined HLS datasource with semantic bands.
Args:
band_names: optional semantic bands to expose.
sources: optional subset of sources to include: sentinel, landsat, or both.
query: optional STAC query dict to include in searches.
sort_by: sort merged STAC results by this property.
sort_ascending: whether the sort should be ascending.
timeout: timeout for auth and asset requests.
earthdata_token: optional Earthdata bearer token override.
s3_credentials_url: LP DAAC temporary credentials endpoint.
context: optional datasource context from rslearn.
"""
if sources is None:
sources = ["sentinel", "landsat"]
normalized_sources: list[str] = []
for source in sources:
if source not in self.SOURCE_TO_COLLECTION:
raise ValueError(
f"unsupported Hls2 source '{source}'. Use one of "
f"{sorted(self.SOURCE_TO_COLLECTION.keys())}."
)
if source not in normalized_sources:
normalized_sources.append(source)
self.sources = normalized_sources
self.collection_names = [
self.SOURCE_TO_COLLECTION[source] for source in self.sources
]
super().__init__(
band_names=band_names,
query=query,
sort_by=sort_by,
sort_ascending=sort_ascending,
timeout=timeout,
earthdata_token=earthdata_token,
s3_credentials_url=s3_credentials_url,
context=context,
collection_name=self.collection_names,
require_asset_filter=False,
)
@override
def _stac_item_to_item(self, stac_item: StacItem) -> SourceItem:
if stac_item.collection is None:
raise ValueError("got unexpected item with no collection")
if stac_item.geometry is None:
raise ValueError("got unexpected item with no geometry")
if stac_item.time_range is None:
raise ValueError("got unexpected item with no time range")
if stac_item.assets is None:
raise ValueError("got unexpected item with no assets")
asset_key_map = self._asset_key_map_for_collection(stac_item.collection)
shp = shapely.geometry.shape(stac_item.geometry)
geometry = STGeometry(WGS84_PROJECTION, shp, stac_item.time_range)
asset_urls = {}
properties: dict[str, Any] = {
"sensor": self.COLLECTION_TO_SOURCE[stac_item.collection],
"collection": stac_item.collection,
}
for semantic_band in self.asset_bands:
stac_asset_key = asset_key_map.get(semantic_band)
if stac_asset_key is None:
continue
s3_asset_key = f"s3_{stac_asset_key}"
if s3_asset_key in stac_item.assets:
asset_urls[semantic_band] = stac_item.assets[s3_asset_key].href
if stac_asset_key in stac_item.assets:
properties[f"{_HTTP_URL_PROPERTY_PREFIX}{semantic_band}"] = (
stac_item.assets[stac_asset_key].href
)
if semantic_band not in asset_urls:
asset_urls[semantic_band] = stac_item.assets[stac_asset_key].href
for prop_name in self.properties_to_record:
if prop_name in stac_item.properties:
properties[prop_name] = stac_item.properties[prop_name]
return SourceItem(stac_item.id, geometry, asset_urls, properties)
@override
def _get_sort_key(self, stac_item: StacItem) -> Any:
assert self.sort_by is not None
return (
stac_item.properties.get(self.sort_by) is None,
stac_item.properties.get(self.sort_by),
)
@override
def _should_include_item(self, item: SourceItem) -> bool:
return all(band in item.asset_urls for band in self.asset_bands)