rslearn.data_sources.planetary_computer

planetary_computer

Data on Planetary Computer.

PlanetaryComputerStacClient

Bases: StacClient

A StacClient subclass that handles Planetary Computer's pagination limits.

Planetary Computer STAC API does not support standard pagination and has a max limit of 1000. If the initial query returns 1000 items, this client paginates by sorting by ID and using gt (greater than) queries to fetch subsequent pages.

Source code in rslearn/data_sources/planetary_computer.py

class PlanetaryComputerStacClient(StacClient):
    """A StacClient subclass that handles Planetary Computer's pagination limits.

    Planetary Computer STAC API does not support standard pagination and has a max
    limit of 1000. If the initial query returns 1000 items, this client paginates
    by sorting by ID and using gt (greater than) queries to fetch subsequent pages.
    """

    @override
    def search(
        self,
        collections: list[str] | None = None,
        bbox: tuple[float, float, float, float] | None = None,
        intersects: dict[str, Any] | None = None,
        date_time: datetime | tuple[datetime, datetime] | None = None,
        ids: list[str] | None = None,
        limit: int | None = None,
        query: dict[str, Any] | None = None,
        sortby: list[dict[str, str]] | None = None,
    ) -> list[StacItem]:
        # We will use sortby for pagination, so the caller must not set it.
        if sortby is not None:
            raise ValueError("sortby must not be set for PlanetaryComputerStacClient")

        # First, try a simple query with the PC limit to detect if pagination is needed.
        # We always use PLANETARY_COMPUTER_LIMIT for the request because PC doesn't
        # support standard pagination, and we need to detect when we hit the limit
        # to switch to ID-based pagination.
        # We could just start sorting by ID here and do pagination, but we treate it as
        # a special case to avoid sorting since that seems to speed up the query.
        stac_items = super().search(
            collections=collections,
            bbox=bbox,
            intersects=intersects,
            date_time=date_time,
            ids=ids,
            limit=PLANETARY_COMPUTER_LIMIT,
            query=query,
        )

        # If we got fewer than the PC limit, we have all the results.
        if len(stac_items) < PLANETARY_COMPUTER_LIMIT:
            return stac_items

        # We hit the limit, so we need to paginate by ID.
        # Re-fetch with sorting by ID to ensure consistent ordering for pagination.
        logger.debug(
            "Initial request returned %d items (at limit), switching to ID pagination",
            len(stac_items),
        )

        all_items: list[StacItem] = []
        last_id: str | None = None

        while True:
            # Build query with id > last_id if we're paginating.
            combined_query: dict[str, Any] = dict(query) if query else {}
            if last_id is not None:
                combined_query["id"] = {"gt": last_id}

            stac_items = super().search(
                collections=collections,
                bbox=bbox,
                intersects=intersects,
                date_time=date_time,
                ids=ids,
                limit=PLANETARY_COMPUTER_LIMIT,
                query=combined_query if combined_query else None,
                sortby=[{"field": "id", "direction": "asc"}],
            )

            all_items.extend(stac_items)

            # If we got fewer than the limit, we've fetched everything.
            if len(stac_items) < PLANETARY_COMPUTER_LIMIT:
                break

            # Otherwise, paginate using the last item's ID.
            last_id = stac_items[-1].id
            logger.debug(
                "Got %d items, paginating with id > %s",
                len(stac_items),
                last_id,
            )

        logger.debug("Total items fetched: %d", len(all_items))
        return all_items

PlanetaryComputer

Bases: DirectMaterializeDataSource[SourceItem], StacDataSource

Modality-agnostic data source for data on Microsoft Planetary Computer.

If there is a subclass available for a modality, it is recommended to use the subclass since it provides additional functionality.

Otherwise, PlanetaryComputer can be configured with the collection name and a dictionary of assets and bands to ingest.

See https://planetarycomputer.microsoft.com/ for details.

The PC_SDK_SUBSCRIPTION_KEY environment variable can be set for higher rate limits but is not needed.

Source code in rslearn/data_sources/planetary_computer.py

class PlanetaryComputer(DirectMaterializeDataSource[SourceItem], StacDataSource):
    """Modality-agnostic data source for data on Microsoft Planetary Computer.

    If there is a subclass available for a modality, it is recommended to use the
    subclass since it provides additional functionality.

    Otherwise, PlanetaryComputer can be configured with the collection name and a
    dictionary of assets and bands to ingest.

    See https://planetarycomputer.microsoft.com/ for details.

    The PC_SDK_SUBSCRIPTION_KEY environment variable can be set for higher rate limits
    but is not needed.
    """

    STAC_ENDPOINT = "https://planetarycomputer.microsoft.com/api/stac/v1"

    def __init__(
        self,
        collection_name: str,
        asset_bands: dict[str, list[str]],
        query: dict[str, Any] | None = None,
        sort_by: str | None = None,
        sort_ascending: bool = True,
        timeout: timedelta = timedelta(seconds=10),
        skip_items_missing_assets: bool = False,
        cache_dir: str | None = None,
        context: DataSourceContext = DataSourceContext(),
    ):
        """Initialize a new PlanetaryComputer instance.

        Args:
            collection_name: the STAC collection name on Planetary Computer.
            asset_bands: assets to ingest, mapping from asset name to the list of bands
                in that asset.
            query: optional query argument to STAC searches.
            sort_by: sort by this property in the STAC items.
            sort_ascending: whether to sort ascending (or descending).
            timeout: timeout for API requests.
            skip_items_missing_assets: skip STAC items that are missing any of the
                assets in asset_bands during get_items.
            cache_dir: deprecated, no longer used. Item data is now passed to
                materialization functions so caching in the data source is unnecessary.
            context: the data source context.
        """
        if cache_dir is not None:
            warnings.warn(
                "cache_dir is deprecated and no longer used. "
                "Item data is now passed directly during materialization.",
                FutureWarning,
                stacklevel=2,
            )

        # Initialize the DirectMaterializeDataSource with asset_bands
        DirectMaterializeDataSource.__init__(self, asset_bands=asset_bands)

        # We pass required_assets to StacDataSource if skip_items_missing_assets is set.
        required_assets: list[str] | None = None
        if skip_items_missing_assets:
            required_assets = list(asset_bands.keys())

        StacDataSource.__init__(
            self,
            endpoint=self.STAC_ENDPOINT,
            collection_name=collection_name,
            query=query,
            sort_by=sort_by,
            sort_ascending=sort_ascending,
            required_assets=required_assets,
        )

        # Replace the client with PlanetaryComputerStacClient to handle PC's pagination limits.
        self.client = PlanetaryComputerStacClient(self.STAC_ENDPOINT)

        self.timeout = timeout
        self.skip_items_missing_assets = skip_items_missing_assets

    # --- DirectMaterializeDataSource implementation ---

    def get_asset_url(self, item: SourceItem, asset_key: str) -> str:
        """Get the signed URL to read the asset for the given item and asset key.

        Args:
            item: the item.
            asset_key: the key identifying which asset to get.

        Returns:
            the signed URL to read the asset from.
        """
        return planetary_computer.sign(item.asset_urls[asset_key])

    def get_raster_bands(self, layer_name: str, item: Item) -> list[list[str]]:
        """Get the sets of bands that have been stored for the specified item.

        Args:
            layer_name: the layer name or alias.
            item: the item.

        Returns:
            a list of lists of bands that are in the tile store (with one raster
                stored corresponding to each inner list). If no rasters are ready for
                this item, returns empty list.
        """
        if not isinstance(item, SourceItem):
            raise TypeError(f"expected SourceItem, got {type(item)}")
        all_bands = []
        for asset_key, band_names in self.asset_bands.items():
            if asset_key not in item.asset_urls:
                continue
            all_bands.append(band_names)
        return all_bands

    # --- DataSource implementation ---

    def ingest(
        self,
        tile_store: TileStoreWithLayer,
        items: list[SourceItem],
        geometries: list[list[STGeometry]],
    ) -> None:
        """Ingest items into the given tile store.

        Args:
            tile_store: the tile store to ingest into
            items: the items to ingest
            geometries: a list of geometries needed for each item
        """
        for item in items:
            for asset_key, band_names in self.asset_bands.items():
                if asset_key not in item.asset_urls:
                    continue
                if tile_store.is_raster_ready(item, band_names):
                    continue

                asset_url = planetary_computer.sign(item.asset_urls[asset_key])

                with tempfile.TemporaryDirectory() as tmp_dir:
                    local_fname = os.path.join(tmp_dir, f"{asset_key}.tif")
                    logger.debug(
                        "PlanetaryComputer download item %s asset %s to %s",
                        item.name,
                        asset_key,
                        local_fname,
                    )
                    with requests.get(
                        asset_url, stream=True, timeout=self.timeout.total_seconds()
                    ) as r:
                        r.raise_for_status()
                        with open(local_fname, "wb") as f:
                            for chunk in r.iter_content(chunk_size=8192):
                                f.write(chunk)

                    logger.debug(
                        "PlanetaryComputer ingest item %s asset %s",
                        item.name,
                        asset_key,
                    )
                    tile_store.write_raster_file(
                        item,
                        band_names,
                        UPath(local_fname),
                        time_range=item.geometry.time_range,
                    )

                logger.debug(
                    "PlanetaryComputer done ingesting item %s asset %s",
                    item.name,
                    asset_key,
                )

get_asset_url

get_asset_url(item: SourceItem, asset_key: str) -> str

Get the signed URL to read the asset for the given item and asset key.

Parameters:

Name	Type	Description	Default
item	SourceItem	the item.	required
asset_key	str	the key identifying which asset to get.	required

Returns:

Type	Description
str	the signed URL to read the asset from.

Source code in rslearn/data_sources/planetary_computer.py

def get_asset_url(self, item: SourceItem, asset_key: str) -> str:
    """Get the signed URL to read the asset for the given item and asset key.

    Args:
        item: the item.
        asset_key: the key identifying which asset to get.

    Returns:
        the signed URL to read the asset from.
    """
    return planetary_computer.sign(item.asset_urls[asset_key])

get_raster_bands

get_raster_bands(layer_name: str, item: Item) -> list[list[str]]

Get the sets of bands that have been stored for the specified item.

Parameters:

Name	Type	Description	Default
layer_name	str	the layer name or alias.	required
item	Item	the item.	required

Returns:

Type	Description
list[list[str]]	a list of lists of bands that are in the tile store (with one raster stored corresponding to each inner list). If no rasters are ready for this item, returns empty list.

Source code in rslearn/data_sources/planetary_computer.py

def get_raster_bands(self, layer_name: str, item: Item) -> list[list[str]]:
    """Get the sets of bands that have been stored for the specified item.

    Args:
        layer_name: the layer name or alias.
        item: the item.

    Returns:
        a list of lists of bands that are in the tile store (with one raster
            stored corresponding to each inner list). If no rasters are ready for
            this item, returns empty list.
    """
    if not isinstance(item, SourceItem):
        raise TypeError(f"expected SourceItem, got {type(item)}")
    all_bands = []
    for asset_key, band_names in self.asset_bands.items():
        if asset_key not in item.asset_urls:
            continue
        all_bands.append(band_names)
    return all_bands

ingest

ingest(tile_store: TileStoreWithLayer, items: list[SourceItem], geometries: list[list[STGeometry]]) -> None

Ingest items into the given tile store.

Parameters:

Name	Type	Description	Default
tile_store	TileStoreWithLayer	the tile store to ingest into	required
items	list[SourceItem]	the items to ingest	required
geometries	list[list[STGeometry]]	a list of geometries needed for each item	required

Source code in rslearn/data_sources/planetary_computer.py

def ingest(
    self,
    tile_store: TileStoreWithLayer,
    items: list[SourceItem],
    geometries: list[list[STGeometry]],
) -> None:
    """Ingest items into the given tile store.

    Args:
        tile_store: the tile store to ingest into
        items: the items to ingest
        geometries: a list of geometries needed for each item
    """
    for item in items:
        for asset_key, band_names in self.asset_bands.items():
            if asset_key not in item.asset_urls:
                continue
            if tile_store.is_raster_ready(item, band_names):
                continue

            asset_url = planetary_computer.sign(item.asset_urls[asset_key])

            with tempfile.TemporaryDirectory() as tmp_dir:
                local_fname = os.path.join(tmp_dir, f"{asset_key}.tif")
                logger.debug(
                    "PlanetaryComputer download item %s asset %s to %s",
                    item.name,
                    asset_key,
                    local_fname,
                )
                with requests.get(
                    asset_url, stream=True, timeout=self.timeout.total_seconds()
                ) as r:
                    r.raise_for_status()
                    with open(local_fname, "wb") as f:
                        for chunk in r.iter_content(chunk_size=8192):
                            f.write(chunk)

                logger.debug(
                    "PlanetaryComputer ingest item %s asset %s",
                    item.name,
                    asset_key,
                )
                tile_store.write_raster_file(
                    item,
                    band_names,
                    UPath(local_fname),
                    time_range=item.geometry.time_range,
                )

            logger.debug(
                "PlanetaryComputer done ingesting item %s asset %s",
                item.name,
                asset_key,
            )

Sentinel2

Bases: PlanetaryComputer

A data source for Sentinel-2 L2A data on Microsoft Planetary Computer.

See https://planetarycomputer.microsoft.com/dataset/sentinel-2-l2a.

Source code in rslearn/data_sources/planetary_computer.py

class Sentinel2(PlanetaryComputer):
    """A data source for Sentinel-2 L2A data on Microsoft Planetary Computer.

    See https://planetarycomputer.microsoft.com/dataset/sentinel-2-l2a.
    """

    COLLECTION_NAME = "sentinel-2-l2a"

    BANDS = {
        "B01": ["B01"],
        "B02": ["B02"],
        "B03": ["B03"],
        "B04": ["B04"],
        "B05": ["B05"],
        "B06": ["B06"],
        "B07": ["B07"],
        "B08": ["B08"],
        "B09": ["B09"],
        "B11": ["B11"],
        "B12": ["B12"],
        "B8A": ["B8A"],
        "SCL": ["SCL"],
        "visual": ["R", "G", "B"],
    }
    NON_REFLECTANCE_ASSETS = frozenset({"SCL", "visual"})

    def __init__(
        self,
        harmonize: bool = False,
        assets: list[str] | None = None,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new Sentinel2 instance.

        Args:
            harmonize: harmonize pixel values across different processing baselines,
                see https://developers.google.com/earth-engine/datasets/catalog/COPERNICUS_S2_SR_HARMONIZED
            assets: list of asset names to ingest, or None to ingest all assets. This
                is only used if the layer config is missing from the context.
            context: the data source context.
            kwargs: other arguments to pass to PlanetaryComputer.
        """
        self.harmonize = harmonize

        # Determine which assets we need based on the bands in the layer config.
        if context.layer_config is not None:
            asset_bands: dict[str, list[str]] = {}
            for asset_key, band_names in self.BANDS.items():
                # See if the bands provided by this asset intersect with the bands in
                # at least one configured band set.
                for band_set in context.layer_config.band_sets:
                    if not set(band_set.bands).intersection(set(band_names)):
                        continue
                    asset_bands[asset_key] = band_names
                    break
        elif assets is not None:
            asset_bands = {asset_key: self.BANDS[asset_key] for asset_key in assets}
        else:
            asset_bands = self.BANDS

        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands=asset_bands,
            # Skip since all of the items should have the same assets.
            skip_items_missing_assets=True,
            context=context,
            **kwargs,
        )

    def _get_product_xml(self, item: SourceItem) -> ET.Element:
        asset_url = planetary_computer.sign(item.asset_urls["product-metadata"])
        response = requests.get(asset_url, timeout=self.timeout.total_seconds())
        response.raise_for_status()
        return ET.fromstring(response.content)

    def _should_harmonize_asset(self, asset_key: str) -> bool:
        """Return whether harmonization should be applied for this Sentinel-2 asset."""
        return self.harmonize and asset_key not in self.NON_REFLECTANCE_ASSETS

    def ingest(
        self,
        tile_store: TileStoreWithLayer,
        items: list[SourceItem],
        geometries: list[list[STGeometry]],
    ) -> None:
        """Ingest items into the given tile store.

        Args:
            tile_store: the tile store to ingest into
            items: the items to ingest
            geometries: a list of geometries needed for each item
        """
        for item in items:
            for asset_key, band_names in self.asset_bands.items():
                if tile_store.is_raster_ready(item, band_names):
                    continue

                asset_url = planetary_computer.sign(item.asset_urls[asset_key])

                with tempfile.TemporaryDirectory() as tmp_dir:
                    local_fname = os.path.join(tmp_dir, f"{asset_key}.tif")
                    logger.debug(
                        "PlanetaryComputer download item %s asset %s to %s",
                        item.name,
                        asset_key,
                        local_fname,
                    )
                    with requests.get(
                        asset_url, stream=True, timeout=self.timeout.total_seconds()
                    ) as r:
                        r.raise_for_status()
                        with open(local_fname, "wb") as f:
                            for chunk in r.iter_content(chunk_size=8192):
                                f.write(chunk)

                    logger.debug(
                        "PlanetaryComputer ingest item %s asset %s",
                        item.name,
                        asset_key,
                    )

                    # Harmonize reflectance assets if needed.
                    harmonize_callback = None
                    if self._should_harmonize_asset(asset_key):
                        harmonize_callback = get_harmonize_callback(
                            self._get_product_xml(item)
                        )

                    if harmonize_callback is not None:
                        # In this case we need to read the array, convert the pixel
                        # values, and pass modified array directly to the TileStore.
                        with rasterio.open(local_fname) as src:
                            array = src.read()
                            src_nodata = src.nodata
                            projection, bounds = get_raster_projection_and_bounds(src)
                        array = harmonize_callback(array)
                        raster_metadata = RasterMetadata(nodata_value=src_nodata)
                        tile_store.write_raster(
                            item,
                            band_names,
                            projection,
                            bounds,
                            RasterArray(
                                chw_array=array,
                                time_range=item.geometry.time_range,
                                metadata=raster_metadata,
                            ),
                        )

                    else:
                        tile_store.write_raster_file(
                            item,
                            band_names,
                            UPath(local_fname),
                            time_range=item.geometry.time_range,
                        )

                logger.debug(
                    "PlanetaryComputer done ingesting item %s asset %s",
                    item.name,
                    asset_key,
                )

    def get_read_callback(
        self, item: SourceItem, asset_key: str
    ) -> Callable[[npt.NDArray[Any]], npt.NDArray[Any]] | None:
        """Return a callback to harmonize Sentinel-2 data if needed.

        Args:
            item: the item being read.
            asset_key: the key identifying which asset is being read.

        Returns:
            A callback function for harmonization, or None if not needed.
        """
        if not self._should_harmonize_asset(asset_key):
            return None

        return get_harmonize_callback(self._get_product_xml(item))

ingest

ingest(tile_store: TileStoreWithLayer, items: list[SourceItem], geometries: list[list[STGeometry]]) -> None

Ingest items into the given tile store.

Parameters:

Name	Type	Description	Default
tile_store	TileStoreWithLayer	the tile store to ingest into	required
items	list[SourceItem]	the items to ingest	required
geometries	list[list[STGeometry]]	a list of geometries needed for each item	required

Source code in rslearn/data_sources/planetary_computer.py

def ingest(
    self,
    tile_store: TileStoreWithLayer,
    items: list[SourceItem],
    geometries: list[list[STGeometry]],
) -> None:
    """Ingest items into the given tile store.

    Args:
        tile_store: the tile store to ingest into
        items: the items to ingest
        geometries: a list of geometries needed for each item
    """
    for item in items:
        for asset_key, band_names in self.asset_bands.items():
            if tile_store.is_raster_ready(item, band_names):
                continue

            asset_url = planetary_computer.sign(item.asset_urls[asset_key])

            with tempfile.TemporaryDirectory() as tmp_dir:
                local_fname = os.path.join(tmp_dir, f"{asset_key}.tif")
                logger.debug(
                    "PlanetaryComputer download item %s asset %s to %s",
                    item.name,
                    asset_key,
                    local_fname,
                )
                with requests.get(
                    asset_url, stream=True, timeout=self.timeout.total_seconds()
                ) as r:
                    r.raise_for_status()
                    with open(local_fname, "wb") as f:
                        for chunk in r.iter_content(chunk_size=8192):
                            f.write(chunk)

                logger.debug(
                    "PlanetaryComputer ingest item %s asset %s",
                    item.name,
                    asset_key,
                )

                # Harmonize reflectance assets if needed.
                harmonize_callback = None
                if self._should_harmonize_asset(asset_key):
                    harmonize_callback = get_harmonize_callback(
                        self._get_product_xml(item)
                    )

                if harmonize_callback is not None:
                    # In this case we need to read the array, convert the pixel
                    # values, and pass modified array directly to the TileStore.
                    with rasterio.open(local_fname) as src:
                        array = src.read()
                        src_nodata = src.nodata
                        projection, bounds = get_raster_projection_and_bounds(src)
                    array = harmonize_callback(array)
                    raster_metadata = RasterMetadata(nodata_value=src_nodata)
                    tile_store.write_raster(
                        item,
                        band_names,
                        projection,
                        bounds,
                        RasterArray(
                            chw_array=array,
                            time_range=item.geometry.time_range,
                            metadata=raster_metadata,
                        ),
                    )

                else:
                    tile_store.write_raster_file(
                        item,
                        band_names,
                        UPath(local_fname),
                        time_range=item.geometry.time_range,
                    )

            logger.debug(
                "PlanetaryComputer done ingesting item %s asset %s",
                item.name,
                asset_key,
            )

get_read_callback

get_read_callback(item: SourceItem, asset_key: str) -> Callable[[NDArray[Any]], NDArray[Any]] | None

Return a callback to harmonize Sentinel-2 data if needed.

Parameters:

Name	Type	Description	Default
item	SourceItem	the item being read.	required
asset_key	str	the key identifying which asset is being read.	required

Returns:

Type	Description
Callable[[NDArray[Any]], NDArray[Any]] | None	A callback function for harmonization, or None if not needed.

Source code in rslearn/data_sources/planetary_computer.py

def get_read_callback(
    self, item: SourceItem, asset_key: str
) -> Callable[[npt.NDArray[Any]], npt.NDArray[Any]] | None:
    """Return a callback to harmonize Sentinel-2 data if needed.

    Args:
        item: the item being read.
        asset_key: the key identifying which asset is being read.

    Returns:
        A callback function for harmonization, or None if not needed.
    """
    if not self._should_harmonize_asset(asset_key):
        return None

    return get_harmonize_callback(self._get_product_xml(item))

Hls2S30

Bases: PlanetaryComputer

A data source for HLS v2 Sentinel-2 (S30) data on Planetary Computer.

Source code in rslearn/data_sources/planetary_computer.py

class Hls2S30(PlanetaryComputer):
    """A data source for HLS v2 Sentinel-2 (S30) data on Planetary Computer."""

    COLLECTION_NAME = "hls2-s30"
    DEFAULT_PLATFORMS = ["sentinel-2a", "sentinel-2b", "sentinel-2c"]
    # Asset keys exposed by the collection.
    ASSET_KEY_TO_COMMON_NAME = {
        "B01": "coastal",
        "B02": "blue",
        "B03": "green",
        "B04": "red",
        "B08": "nir",
        "B10": "cirrus",
        "B11": "swir16",
        "B12": "swir22",
    }
    COMMON_NAME_TO_ASSET_KEY = {
        common: asset for asset, common in ASSET_KEY_TO_COMMON_NAME.items()
    }
    DEFAULT_BANDS = list(ASSET_KEY_TO_COMMON_NAME.keys())

    @classmethod
    def _normalize_band_name(cls, band: str) -> str:
        if band in cls.ASSET_KEY_TO_COMMON_NAME:
            return band
        if band in cls.COMMON_NAME_TO_ASSET_KEY:
            return cls.COMMON_NAME_TO_ASSET_KEY[band]
        raise ValueError(
            f"unsupported HLS2 S30 band '{band}'. Use one of {sorted(cls.ASSET_KEY_TO_COMMON_NAME.keys())} "
            f"(asset keys) or {sorted(cls.COMMON_NAME_TO_ASSET_KEY.keys())} (common names)."
        )

    def __init__(
        self,
        band_names: list[str] | None = None,
        platforms: list[str] | None = None,
        query: dict[str, Any] | None = None,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new Hls2S30 instance.

        Args:
            band_names: optional list of bands to expose. If not provided and a layer
                config is present, bands are inferred from the band sets. Otherwise
                defaults to the HLS S30 reflectance bands.
            platforms: optional list of Sentinel-2 platform identifiers to include.
                Defaults to ["sentinel-2a", "sentinel-2b", "sentinel-2c"].
            query: optional STAC query filter to use. If not set, this defaults to a
                platform filter for the configured platforms.
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        if context.layer_config is not None:
            requested_bands = {
                band
                for band_set in context.layer_config.band_sets
                for band in band_set.bands
            }
            band_names = [self._normalize_band_name(band) for band in requested_bands]
        elif band_names is None:
            band_names = self.DEFAULT_BANDS
        else:
            band_names = [self._normalize_band_name(band) for band in band_names]

        if platforms is None:
            platforms = self.DEFAULT_PLATFORMS

        if query is None:
            query = {"platform": {"in": platforms}}

        # Assets are keyed by band name; each asset is a single band.
        asset_bands = {band: [band] for band in band_names}

        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands=asset_bands,
            query=query,
            # Skip per-item asset checks; required assets are derived from asset_bands.
            skip_items_missing_assets=True,
            context=context,
            **kwargs,
        )

Hls2L30

Bases: PlanetaryComputer

A data source for HLS v2 Landsat (L30) data on Planetary Computer.

Source code in rslearn/data_sources/planetary_computer.py

class Hls2L30(PlanetaryComputer):
    """A data source for HLS v2 Landsat (L30) data on Planetary Computer."""

    COLLECTION_NAME = "hls2-l30"
    DEFAULT_PLATFORMS = ["landsat-8", "landsat-9"]
    ASSET_KEY_TO_COMMON_NAME = {
        "B01": "coastal",
        "B02": "blue",
        "B03": "green",
        "B04": "red",
        "B05": "nir",
        "B06": "swir16",
        "B07": "swir22",
        "B09": "cirrus",
        "B10": "lwir11",
        "B11": "lwir12",
    }
    COMMON_NAME_TO_ASSET_KEY = {
        common: asset for asset, common in ASSET_KEY_TO_COMMON_NAME.items()
    }
    DEFAULT_BANDS = list(ASSET_KEY_TO_COMMON_NAME.keys())

    @classmethod
    def _normalize_band_name(cls, band: str) -> str:
        if band in cls.ASSET_KEY_TO_COMMON_NAME:
            return band
        if band in cls.COMMON_NAME_TO_ASSET_KEY:
            return cls.COMMON_NAME_TO_ASSET_KEY[band]
        raise ValueError(
            f"unknown HLS2 L30 band '{band}'. Use one of {sorted(cls.ASSET_KEY_TO_COMMON_NAME.keys())} "
            f"(asset keys) or {sorted(cls.COMMON_NAME_TO_ASSET_KEY.keys())} (common names)."
        )

    def __init__(
        self,
        band_names: list[str] | None = None,
        platforms: list[str] | None = None,
        query: dict[str, Any] | None = None,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new Hls2L30 instance.

        Args:
            band_names: optional list of bands to expose. If not provided and a layer
                config is present, bands are inferred from the band sets. Otherwise
                defaults to the HLS L30 reflectance bands.
            platforms: optional list of Landsat platform identifiers to include.
                Defaults to ["landsat-8", "landsat-9"].
            query: optional STAC query filter to use. If not set, this defaults to a
                platform filter for the configured platforms.
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        if context.layer_config is not None:
            requested_bands = {
                band
                for band_set in context.layer_config.band_sets
                for band in band_set.bands
            }
            band_names = [self._normalize_band_name(band) for band in requested_bands]
        elif band_names is None:
            band_names = self.DEFAULT_BANDS
        else:
            band_names = [self._normalize_band_name(band) for band in band_names]

        if platforms is None:
            platforms = self.DEFAULT_PLATFORMS

        if query is None:
            query = {"platform": {"in": platforms}}

        asset_bands = {band: [band] for band in band_names}

        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands=asset_bands,
            query=query,
            # Skip per-item asset checks; required assets are derived from asset_bands.
            skip_items_missing_assets=True,
            context=context,
            **kwargs,
        )

LandsatC2L2

Bases: PlanetaryComputer

A data source for Landsat Collection 2 Level-2 data on Planetary Computer.

This data source targets Landsat 8/9 items in the landsat-c2-l2 collection. Band names exposed by this data source are Landsat-style band identifiers (e.g. "B4", "B5", "B10") for maximum compatibility with rslearn.data_sources.aws_landsat.LandsatOliTirs.

For convenience, configuration also accepts STAC common_name values (e.g. "red", "nir08") and STAC eo:bands[].name aliases (e.g. "OLI_B4", "TIRS_B10"), which are normalized to the Landsat-style band identifiers above.

Note: this is Level-2 data, not Level-1. If you need Level-1-specific bands (e.g. panchromatic/cirrus or thermal band 11), use rslearn.data_sources.aws_landsat.LandsatOliTirs.

Source code in rslearn/data_sources/planetary_computer.py

class LandsatC2L2(PlanetaryComputer):
    """A data source for Landsat Collection 2 Level-2 data on Planetary Computer.

    This data source targets Landsat 8/9 items in the `landsat-c2-l2` collection.
    Band names exposed by this data source are Landsat-style band identifiers
    (e.g. "B4", "B5", "B10") for maximum compatibility with
    `rslearn.data_sources.aws_landsat.LandsatOliTirs`.

    For convenience, configuration also accepts STAC `common_name` values (e.g. "red",
    "nir08") and STAC `eo:bands[].name` aliases (e.g. "OLI_B4", "TIRS_B10"), which are
    normalized to the Landsat-style band identifiers above.

    Note: this is Level-2 data, not Level-1. If you need Level-1-specific bands
    (e.g. panchromatic/cirrus or thermal band 11), use
    `rslearn.data_sources.aws_landsat.LandsatOliTirs`.
    """

    COLLECTION_NAME = "landsat-c2-l2"

    # Map STAC asset keys (common_name) to the Landsat band identifiers we expose.
    # Planetary Computer assets for `landsat-c2-l2` are keyed by common_name.
    ASSET_COMMON_NAME_TO_BAND = {
        "coastal": "B1",
        "blue": "B2",
        "green": "B3",
        "red": "B4",
        "nir08": "B5",
        "swir16": "B6",
        "swir22": "B7",
        "lwir11": "B10",
    }

    BAND_TO_ASSET_COMMON_NAME = {v: k for k, v in ASSET_COMMON_NAME_TO_BAND.items()}

    # STAC eo:bands name -> Landsat-style band identifiers.
    STAC_BAND_NAME_ALIASES = {
        "OLI_B1": "B1",
        "OLI_B2": "B2",
        "OLI_B3": "B3",
        "OLI_B4": "B4",
        "OLI_B5": "B5",
        "OLI_B6": "B6",
        "OLI_B7": "B7",
        "TIRS_B10": "B10",
    }

    DEFAULT_PLATFORM_QUERY = {"platform": {"in": ["landsat-8", "landsat-9"]}}

    @classmethod
    def _normalize_band_name(cls, band: str) -> str:
        if band in cls.BAND_TO_ASSET_COMMON_NAME:
            return band
        if band in cls.ASSET_COMMON_NAME_TO_BAND:
            return cls.ASSET_COMMON_NAME_TO_BAND[band]
        if band in cls.STAC_BAND_NAME_ALIASES:
            return cls.STAC_BAND_NAME_ALIASES[band]
        if band in {"B8", "B9", "B11"}:
            raise ValueError(
                f"LandsatC2L2 does not provide {band} in the Planetary Computer "
                "landsat-c2-l2 collection. Use rslearn.data_sources.aws_landsat.LandsatOliTirs "
                "for Level-1 bands like panchromatic (B8), cirrus (B9), or thermal band 11 (B11)."
            )
        raise ValueError(
            f"unknown Landsat band '{band}'. Use one of {sorted(cls.BAND_TO_ASSET_COMMON_NAME.keys())} "
            f"(Landsat band names), {sorted(cls.ASSET_COMMON_NAME_TO_BAND.keys())} (STAC common names), "
            f"or {sorted(cls.STAC_BAND_NAME_ALIASES.keys())} (STAC band names)."
        )

    def __init__(
        self,
        band_names: list[str] | None = None,
        query: dict[str, Any] | None = None,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new LandsatC2L2 instance.

        Args:
            band_names: optional list of band names to expose. Values can be either
                STAC common names (preferred) or STAC `eo:bands[].name` aliases.
                If not provided, defaults to the reflectance bands listed in BANDS.
            query: optional STAC query filter to use. If not set, this defaults to a
                platform filter for Landsat 8/9. If set, the provided query is used
                as-is (no implicit platform filtering is added).
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        # Prefer determining bands from the configured layer config (if present).
        if context.layer_config is not None:
            requested_bands = {
                band
                for band_set in context.layer_config.band_sets
                for band in band_set.bands
            }
            band_names = [self._normalize_band_name(band) for band in requested_bands]
        elif band_names is not None:
            band_names = [self._normalize_band_name(band) for band in band_names]
        else:
            band_names = list(self.BAND_TO_ASSET_COMMON_NAME.keys())

        # Landsat C2 L2 assets are keyed by common name; each asset is a single band.
        # We expose Landsat-style band identifiers (B1, B2, ...).
        asset_bands = {
            self.BAND_TO_ASSET_COMMON_NAME[band]: [band] for band in band_names
        }

        if query is None:
            query = self.DEFAULT_PLATFORM_QUERY

        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands=asset_bands,
            query=query,
            # Skip per-item asset checks; required assets are derived from asset_bands.
            skip_items_missing_assets=True,
            context=context,
            **kwargs,
        )

Sentinel1

Bases: PlanetaryComputer

A data source for Sentinel-1 data on Microsoft Planetary Computer.

This uses the radiometrically corrected data.

See https://planetarycomputer.microsoft.com/dataset/sentinel-1-rtc.

Source code in rslearn/data_sources/planetary_computer.py

class Sentinel1(PlanetaryComputer):
    """A data source for Sentinel-1 data on Microsoft Planetary Computer.

    This uses the radiometrically corrected data.

    See https://planetarycomputer.microsoft.com/dataset/sentinel-1-rtc.
    """

    COLLECTION_NAME = "sentinel-1-rtc"

    def __init__(
        self,
        band_names: list[str] | None = None,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new Sentinel1 instance.

        Args:
            band_names: list of bands to try to ingest, if the layer config is missing
                from the context.
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        # Get band names from the config if possible. If it isn't in the context, then
        # we have to use the provided band names.
        if context.layer_config is not None:
            band_names = list(
                {
                    band
                    for band_set in context.layer_config.band_sets
                    for band in band_set.bands
                }
            )
        if band_names is None:
            raise ValueError(
                "band_names must be set if layer config is not in the context"
            )
        # For Sentinel-1, the asset key should be the same as the band name (and all
        # assets have one band).
        asset_bands = {band: [band] for band in band_names}
        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands=asset_bands,
            context=context,
            **kwargs,
        )

Naip

Bases: PlanetaryComputer

A data source for NAIP data on Microsoft Planetary Computer.

See https://planetarycomputer.microsoft.com/dataset/naip.

Source code in rslearn/data_sources/planetary_computer.py

class Naip(PlanetaryComputer):
    """A data source for NAIP data on Microsoft Planetary Computer.

    See https://planetarycomputer.microsoft.com/dataset/naip.
    """

    COLLECTION_NAME = "naip"
    ASSET_BANDS = {"image": ["R", "G", "B", "NIR"]}

    def __init__(
        self,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new Naip instance.

        Args:
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands=self.ASSET_BANDS,
            context=context,
            **kwargs,
        )

CopDemGlo30

Bases: PlanetaryComputer

A data source for Copernicus DEM GLO-30 (30m) on Microsoft Planetary Computer.

See https://planetarycomputer.microsoft.com/dataset/cop-dem-glo-30.

Source code in rslearn/data_sources/planetary_computer.py

class CopDemGlo30(PlanetaryComputer):
    """A data source for Copernicus DEM GLO-30 (30m) on Microsoft Planetary Computer.

    See https://planetarycomputer.microsoft.com/dataset/cop-dem-glo-30.
    """

    COLLECTION_NAME = "cop-dem-glo-30"
    DATA_ASSET = "data"

    def __init__(
        self,
        band_name: str = "DEM",
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ):
        """Initialize a new CopDemGlo30 instance.

        Args:
            band_name: band name to use if the layer config is missing from the
                context.
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        if context.layer_config is not None:
            if len(context.layer_config.band_sets) != 1:
                raise ValueError("expected a single band set")
            if len(context.layer_config.band_sets[0].bands) != 1:
                raise ValueError("expected band set to have a single band")
            band_name = context.layer_config.band_sets[0].bands[0]

        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands={self.DATA_ASSET: [band_name]},
            # Skip since all items should have the same asset(s).
            skip_items_missing_assets=True,
            context=context,
            **kwargs,
        )

    def _stac_item_to_item(self, stac_item: Any) -> SourceItem:
        # Copernicus DEM is static; ignore item timestamps so it matches any window.
        item = super()._stac_item_to_item(stac_item)
        item.geometry = STGeometry(item.geometry.projection, item.geometry.shp, None)
        return item

    def _get_search_time_range(self, geometry: STGeometry) -> None:
        # Copernicus DEM is static; do not filter STAC searches by time.
        return None

Sentinel3SlstrLST

Bases: PlanetaryComputer

Sentinel-3 SLSTR L2 Land Surface Temperature data on Planetary Computer.

This collection provides netCDF swaths with geolocation arrays. We interpolate the swath onto a regular lat/lon grid using linear interpolation during ingestion. Direct materialization is not supported; keep ingest enabled.

Requires the optional netCDF/xarray dependencies (netCDF4/h5netcdf/h5py).

Source code in rslearn/data_sources/planetary_computer.py

class Sentinel3SlstrLST(PlanetaryComputer):
    """Sentinel-3 SLSTR L2 Land Surface Temperature data on Planetary Computer.

    This collection provides netCDF swaths with geolocation arrays. We interpolate
    the swath onto a regular lat/lon grid using linear interpolation during ingestion.
    Direct materialization is not supported; keep ingest enabled.

    Requires the optional netCDF/xarray dependencies (netCDF4/h5netcdf/h5py).
    """

    COLLECTION_NAME = "sentinel-3-slstr-lst-l2-netcdf"
    LST_ASSET_KEY = "lst-in"
    GEODETIC_ASSET_KEY = "slstr-geodetic-in"
    DEFAULT_BANDS = ["LST"]

    def __init__(
        self,
        sample_step: int = 20,
        nodata_value: float = 0.0,
        grid_resolution: float | None = None,
        context: DataSourceContext = DataSourceContext(),
        **kwargs: Any,
    ) -> None:
        """Initialize a new Sentinel3SlstrLST instance.

        Args:
            sample_step: stride (in pixels) for sampling the geodetic arrays when
                estimating grid resolution.
            nodata_value: value to use for missing data in the output GeoTIFF.
            grid_resolution: optional output grid resolution (degrees). If not set,
                it is estimated from the geodetic arrays.
            context: the data source context.
            kwargs: additional arguments to pass to PlanetaryComputer.
        """
        self.sample_step = max(1, sample_step)
        self.nodata_value = nodata_value
        self.grid_resolution = grid_resolution

        if context.layer_config is not None:
            requested_bands = {
                band
                for band_set in context.layer_config.band_sets
                for band in band_set.bands
            }
            if requested_bands != set(self.DEFAULT_BANDS):
                raise ValueError(
                    "Sentinel3SlstrLST only supports the LST band. "
                    f"Requested: {sorted(requested_bands)}"
                )

        self.band_names = self.DEFAULT_BANDS

        super().__init__(
            collection_name=self.COLLECTION_NAME,
            asset_bands={self.LST_ASSET_KEY: self.band_names},
            skip_items_missing_assets=True,
            context=context,
            **kwargs,
        )

    def _estimate_grid_resolution(
        self, lons: npt.NDArray[np.floating], lats: npt.NDArray[np.floating]
    ) -> float:
        """Estimate grid resolution in degrees from geodetic arrays."""
        if lons.shape != lats.shape:
            raise ValueError(
                f"expected lon/lat arrays to have same shape, got {lons.shape} and {lats.shape}"
            )
        step = max(1, self.sample_step)
        lons_s = lons[::step, ::step]
        lats_s = lats[::step, ::step]

        lon_diff = np.abs(np.diff(lons_s, axis=1)).ravel()
        lat_diff = np.abs(np.diff(lats_s, axis=0)).ravel()
        diffs = np.concatenate([lon_diff, lat_diff])
        diffs = diffs[np.isfinite(diffs) & (diffs > 0)]
        if diffs.size == 0:
            return 0.01
        return float(np.median(diffs))

    def _mask_geodetic_by_valid_data(
        self,
        lons: npt.NDArray[np.floating],
        lats: npt.NDArray[np.floating],
        data: npt.NDArray[np.floating],
    ) -> tuple[npt.NDArray[np.floating], npt.NDArray[np.floating]]:
        """Mask lon/lat arrays where the data is invalid."""
        valid_mask = np.isfinite(data[0])
        lons = np.where(valid_mask, lons, np.nan)
        lats = np.where(valid_mask, lats, np.nan)
        return lons, lats

    def ingest(
        self,
        tile_store: TileStoreWithLayer,
        items: list[SourceItem],
        geometries: list[list[STGeometry]],
    ) -> None:
        """Ingest items into the given tile store."""
        for item in items:
            if tile_store.is_raster_ready(item, self.band_names):
                continue

            if self.LST_ASSET_KEY not in item.asset_urls:
                logger.warning(
                    "Sentinel3SlstrLST item %s missing asset %s, skipping",
                    item.name,
                    self.LST_ASSET_KEY,
                )
                continue
            if self.GEODETIC_ASSET_KEY not in item.asset_urls:
                logger.warning(
                    "Sentinel3SlstrLST item %s missing asset %s, skipping",
                    item.name,
                    self.GEODETIC_ASSET_KEY,
                )
                continue

            lst_url = planetary_computer.sign(item.asset_urls[self.LST_ASSET_KEY])
            geodetic_url = planetary_computer.sign(
                item.asset_urls[self.GEODETIC_ASSET_KEY]
            )

            with tempfile.TemporaryDirectory() as tmp_dir:
                lst_path = os.path.join(tmp_dir, "lst-in.nc")
                geodetic_path = os.path.join(tmp_dir, "geodetic-in.nc")
                for url, path in ((lst_url, lst_path), (geodetic_url, geodetic_path)):
                    with requests.get(
                        url, stream=True, timeout=self.timeout.total_seconds()
                    ) as r:
                        r.raise_for_status()
                        with open(path, "wb") as f:
                            for chunk in r.iter_content(chunk_size=8192):
                                f.write(chunk)

                with (
                    xr.open_dataset(lst_path, mask_and_scale=True) as lst_ds,
                    xr.open_dataset(geodetic_path, mask_and_scale=True) as geo_ds,
                ):
                    lons = np.asarray(geo_ds["longitude_in"].values, dtype=np.float64)
                    lats = np.asarray(geo_ds["latitude_in"].values, dtype=np.float64)

                    band_arrays = []
                    for band in self.band_names:
                        if band not in lst_ds:
                            raise ValueError(
                                f"Sentinel3SlstrLST band '{band}' not found in {self.LST_ASSET_KEY}"
                            )
                        band_arrays.append(
                            np.asarray(lst_ds[band].values, dtype=np.float32)
                        )

                    stack = np.stack(band_arrays, axis=0)
                    lons, lats = self._mask_geodetic_by_valid_data(lons, lats, stack)

                    grid_resolution = (
                        self.grid_resolution
                        if self.grid_resolution is not None
                        else self._estimate_grid_resolution(lons, lats)
                    )
                    logger.debug(
                        "SLSTR LST grid resolution (deg): %s",
                        grid_resolution,
                    )
                    gridded_array, projection, bounds = interpolate_to_grid(
                        data=stack,
                        lon=lons,
                        lat=lats,
                        grid_resolution=grid_resolution,
                    )

                    if self.nodata_value != NODATA_VALUE:
                        gridded_array = np.where(
                            gridded_array == NODATA_VALUE,
                            self.nodata_value,
                            gridded_array,
                        )

                raster_metadata = RasterMetadata(nodata_value=self.nodata_value)
                tile_store.write_raster(
                    item,
                    self.band_names,
                    projection,
                    bounds,
                    RasterArray(
                        chw_array=gridded_array,
                        time_range=item.geometry.time_range,
                        metadata=raster_metadata,
                    ),
                )

    def read_raster(
        self,
        layer_name: str,
        item: Item,
        bands: list[str],
        projection: Any,
        bounds: Any,
        resampling: Any = rasterio.enums.Resampling.bilinear,
    ) -> RasterArray:
        """Direct materialization is not supported for this data source."""
        raise NotImplementedError(
            "Sentinel3SlstrLST does not support direct materialization; set ingest=true."
        )

ingest

ingest(tile_store: TileStoreWithLayer, items: list[SourceItem], geometries: list[list[STGeometry]]) -> None

Ingest items into the given tile store.

Source code in rslearn/data_sources/planetary_computer.py

def ingest(
    self,
    tile_store: TileStoreWithLayer,
    items: list[SourceItem],
    geometries: list[list[STGeometry]],
) -> None:
    """Ingest items into the given tile store."""
    for item in items:
        if tile_store.is_raster_ready(item, self.band_names):
            continue

        if self.LST_ASSET_KEY not in item.asset_urls:
            logger.warning(
                "Sentinel3SlstrLST item %s missing asset %s, skipping",
                item.name,
                self.LST_ASSET_KEY,
            )
            continue
        if self.GEODETIC_ASSET_KEY not in item.asset_urls:
            logger.warning(
                "Sentinel3SlstrLST item %s missing asset %s, skipping",
                item.name,
                self.GEODETIC_ASSET_KEY,
            )
            continue

        lst_url = planetary_computer.sign(item.asset_urls[self.LST_ASSET_KEY])
        geodetic_url = planetary_computer.sign(
            item.asset_urls[self.GEODETIC_ASSET_KEY]
        )

        with tempfile.TemporaryDirectory() as tmp_dir:
            lst_path = os.path.join(tmp_dir, "lst-in.nc")
            geodetic_path = os.path.join(tmp_dir, "geodetic-in.nc")
            for url, path in ((lst_url, lst_path), (geodetic_url, geodetic_path)):
                with requests.get(
                    url, stream=True, timeout=self.timeout.total_seconds()
                ) as r:
                    r.raise_for_status()
                    with open(path, "wb") as f:
                        for chunk in r.iter_content(chunk_size=8192):
                            f.write(chunk)

            with (
                xr.open_dataset(lst_path, mask_and_scale=True) as lst_ds,
                xr.open_dataset(geodetic_path, mask_and_scale=True) as geo_ds,
            ):
                lons = np.asarray(geo_ds["longitude_in"].values, dtype=np.float64)
                lats = np.asarray(geo_ds["latitude_in"].values, dtype=np.float64)

                band_arrays = []
                for band in self.band_names:
                    if band not in lst_ds:
                        raise ValueError(
                            f"Sentinel3SlstrLST band '{band}' not found in {self.LST_ASSET_KEY}"
                        )
                    band_arrays.append(
                        np.asarray(lst_ds[band].values, dtype=np.float32)
                    )

                stack = np.stack(band_arrays, axis=0)
                lons, lats = self._mask_geodetic_by_valid_data(lons, lats, stack)

                grid_resolution = (
                    self.grid_resolution
                    if self.grid_resolution is not None
                    else self._estimate_grid_resolution(lons, lats)
                )
                logger.debug(
                    "SLSTR LST grid resolution (deg): %s",
                    grid_resolution,
                )
                gridded_array, projection, bounds = interpolate_to_grid(
                    data=stack,
                    lon=lons,
                    lat=lats,
                    grid_resolution=grid_resolution,
                )

                if self.nodata_value != NODATA_VALUE:
                    gridded_array = np.where(
                        gridded_array == NODATA_VALUE,
                        self.nodata_value,
                        gridded_array,
                    )

            raster_metadata = RasterMetadata(nodata_value=self.nodata_value)
            tile_store.write_raster(
                item,
                self.band_names,
                projection,
                bounds,
                RasterArray(
                    chw_array=gridded_array,
                    time_range=item.geometry.time_range,
                    metadata=raster_metadata,
                ),
            )

read_raster

read_raster(layer_name: str, item: Item, bands: list[str], projection: Any, bounds: Any, resampling: Any = bilinear) -> RasterArray

Direct materialization is not supported for this data source.

Source code in rslearn/data_sources/planetary_computer.py

def read_raster(
    self,
    layer_name: str,
    item: Item,
    bands: list[str],
    projection: Any,
    bounds: Any,
    resampling: Any = rasterio.enums.Resampling.bilinear,
) -> RasterArray:
    """Direct materialization is not supported for this data source."""
    raise NotImplementedError(
        "Sentinel3SlstrLST does not support direct materialization; set ingest=true."
    )