sentinel_3

Sentinel3Search

Sentinel3Search(
    collection: PlanetaryComputerS3Collection | str = OLCI_WFR,
    date_range: DateLike = None,
    bbox: BBoxLike | None = None,
    intersects: IntersectsLike | None = None,
    logger: Logger = LOGGER,
)

Bases: AbstractStacWrapper

Executable wrapper for Sentinel-3 OLCI data on Planetary Computer.

Implements a fluent builder pattern to construct STAC queries. Execution and result storage are delegated to an underlying StacSearch client via proxy properties.

Initialize Sentinel3Search.

Parameters:

Name	Type	Description	Default
collection	PlanetaryComputerS3Collection | str	The Sentinel-3 STAC collection (default: sentinel-3-olci-wfr-l2-netcdf).	OLCI_WFR
date_range	DateLike	Temporal filter, native pystac DateLike.	None
bbox	BBoxLike | None	Spatial bounding box filter.	None
intersects	IntersectsLike | None	Spatial GeoJSON geometry filter.	None
logger	Logger	Custom logger instance.	LOGGER

Source code in src/geospatial_tools/stac/planetary_computer/sentinel_3.py

def __init__(
    self,
    collection: PlanetaryComputerS3Collection | str = PlanetaryComputerS3Collection.OLCI_WFR,
    date_range: DateLike = None,
    bbox: BBoxLike | None = None,
    intersects: IntersectsLike | None = None,
    logger: logging.Logger = LOGGER,
) -> None:
    """
    Initialize Sentinel3Search.

    Args:
        collection: The Sentinel-3 STAC collection (default: sentinel-3-olci-wfr-l2-netcdf).
        date_range: Temporal filter, native pystac DateLike.
        bbox: Spatial bounding box filter.
        intersects: Spatial GeoJSON geometry filter.
        logger: Custom logger instance.
    """
    super().__init__(collection=collection, date_range=date_range, bbox=bbox, intersects=intersects, logger=logger)

    self.orbit_states: list[PlanetaryComputerS3OrbitState] | None = None
    self.custom_query_params: dict[str, Any] = {}

search_results property

search_results: list[Item] | None

Proxy property for STAC search results from the underlying client.

downloaded_assets property

downloaded_assets: list[Asset] | None

Proxy property for downloaded assets from the underlying client.

filter_by_orbit_state

filter_by_orbit_state(
    states: list[PlanetaryComputerS3OrbitState] | PlanetaryComputerS3OrbitState,
) -> Self

Filter products by orbit state (ascending or descending).

Invalidates current search results.

Parameters:

Name	Type	Description	Default
states	list[PlanetaryComputerS3OrbitState] | PlanetaryComputerS3OrbitState	Single state or list of PlanetaryComputerS3OrbitState.	required

Returns:

Type	Description
Self	The instance itself (Self) for fluent chaining.

Source code in src/geospatial_tools/stac/planetary_computer/sentinel_3.py

def filter_by_orbit_state(
    self, states: list[PlanetaryComputerS3OrbitState] | PlanetaryComputerS3OrbitState
) -> Self:
    """
    Filter products by orbit state (ascending or descending).

    Invalidates current search results.

    Args:
        states: Single state or list of `PlanetaryComputerS3OrbitState`.

    Returns:
        The instance itself (Self) for fluent chaining.
    """
    self._invalidate_state()
    if isinstance(states, list):
        self.orbit_states = states
    else:
        self.orbit_states = [states]
    return self

download

download(
    bands: list[PlanetaryComputerS3Band | str], base_directory: str | Path
) -> list[Asset] | None

Download Sentinel-3 assets with lowercase band key normalization.

Parameters:

Name	Type	Description	Default
bands	list[PlanetaryComputerS3Band | str]	List of bands to download.	required
base_directory	str | Path	Local directory where assets will be saved.	required

Returns:

Type	Description
list[Asset] | None	List of downloaded Asset objects.

Source code in src/geospatial_tools/stac/planetary_computer/sentinel_3.py

def download(self, bands: list[PlanetaryComputerS3Band | str], base_directory: str | Path) -> list[Asset] | None:
    """
    Download Sentinel-3 assets with lowercase band key normalization.

    Args:
        bands: List of bands to download.
        base_directory: Local directory where assets will be saved.

    Returns:
        List of downloaded Asset objects.
    """
    lower_bands = [str(b).lower() for b in bands]
    return super().download(bands=lower_bands, base_directory=base_directory)

with_custom_query

with_custom_query(query_params: dict[str, Any]) -> Self

Merge custom STAC query parameters and invalidate current state.

Parameters:

Name	Type	Description	Default
query_params	dict[str, Any]	Dictionary of custom STAC query parameters.	required

Returns:

Type	Description
Self	The instance itself (Self) for fluent chaining.

Source code in src/geospatial_tools/stac/core.py

def with_custom_query(self, query_params: dict[str, Any]) -> Self:
    """
    Merge custom STAC query parameters and invalidate current state.

    Args:
        query_params: Dictionary of custom STAC query parameters.

    Returns:
        The instance itself (Self) for fluent chaining.
    """
    self._invalidate_state()
    self.custom_query_params.update(query_params)
    return self

search

search() -> list[Item] | None

Execute the STAC search using the built query and parameters.

Returns:

Type	Description
list[Item] | None	List of matched pystac Items, or None if no results.

Source code in src/geospatial_tools/stac/core.py

def search(self) -> list[pystac.Item] | None:
    """
    Execute the STAC search using the built query and parameters.

    Returns:
        List of matched pystac Items, or None if no results.
    """
    query = self._build_collection_query()
    query.update(self.custom_query_params)

    collections = [self.collection] if self.collection else None

    self.client.search(
        collections=collections,
        bbox=self.bbox,
        intersects=self.intersects,
        date_range=self.date_range,
        query=query if query else None,
    )
    return self.search_results