stac module

PlanetaryComputerEndpoint

Bases: TitilerEndpoint

This class contains the methods for the Microsoft Planetary Computer endpoint.

Source code in leafmap/stac.py

class PlanetaryComputerEndpoint(TitilerEndpoint):
    """This class contains the methods for the Microsoft Planetary Computer endpoint."""

    def __init__(
        self,
        endpoint: Optional[str] = "https://planetarycomputer.microsoft.com/api/data/v1",
        name: Optional[str] = "item",
        TileMatrixSetId: Optional[str] = "WebMercatorQuad",
    ):
        """Initialize the PlanetaryComputerEndpoint object.

        Args:
            endpoint (str, optional): The endpoint of the titiler server. Defaults to "https://planetarycomputer.microsoft.com/api/data/v1".
            name (str, optional): The name to be used in the file path. Defaults to "item".
            TileMatrixSetId (str, optional): The TileMatrixSetId to be used in the file path. Defaults to "WebMercatorQuad".
        """
        super().__init__(endpoint, name, TileMatrixSetId)

    def url_for_stac_collection(self):
        return f"{self.endpoint}/collection/{self.TileMatrixSetId}/tilejson.json"

    def url_for_collection_assets(self):
        return f"{self.endpoint}/collection/assets"

    def url_for_collection_bounds(self):
        return f"{self.endpoint}/collection/bounds"

    def url_for_collection_info(self):
        return f"{self.endpoint}/collection/info"

    def url_for_collection_info_geojson(self):
        return f"{self.endpoint}/collection/info.geojson"

    def url_for_collection_pixel_value(self, lon, lat):
        return f"{self.endpoint}/collection/point/{lon},{lat}"

    def url_for_collection_wmts(self):
        return f"{self.endpoint}/collection/{self.TileMatrixSetId}/WMTSCapabilities.xml"

    def url_for_collection_lat_lon_assets(self, lng, lat):
        return f"{self.endpoint}/collection/{lng},{lat}/assets"

    def url_for_collection_bbox_assets(self, minx, miny, maxx, maxy):
        return f"{self.endpoint}/collection/{minx},{miny},{maxx},{maxy}/assets"

    def url_for_stac_mosaic(self, searchid):
        return f"{self.endpoint}/mosaic/{searchid}/{self.TileMatrixSetId}/tilejson.json"

    def url_for_mosaic_info(self, searchid):
        return f"{self.endpoint}/mosaic/{searchid}/info"

    def url_for_mosaic_lat_lon_assets(self, searchid, lon, lat):
        return f"{self.endpoint}/mosaic/{searchid}/{lon},{lat}/assets"

__init__(endpoint='https://planetarycomputer.microsoft.com/api/data/v1', name='item', TileMatrixSetId='WebMercatorQuad')

Initialize the PlanetaryComputerEndpoint object.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint of the titiler server. Defaults to "https://planetarycomputer.microsoft.com/api/data/v1".	'https://planetarycomputer.microsoft.com/api/data/v1'
name	str	The name to be used in the file path. Defaults to "item".	'item'
TileMatrixSetId	str	The TileMatrixSetId to be used in the file path. Defaults to "WebMercatorQuad".	'WebMercatorQuad'

Source code in leafmap/stac.py

def __init__(
    self,
    endpoint: Optional[str] = "https://planetarycomputer.microsoft.com/api/data/v1",
    name: Optional[str] = "item",
    TileMatrixSetId: Optional[str] = "WebMercatorQuad",
):
    """Initialize the PlanetaryComputerEndpoint object.

    Args:
        endpoint (str, optional): The endpoint of the titiler server. Defaults to "https://planetarycomputer.microsoft.com/api/data/v1".
        name (str, optional): The name to be used in the file path. Defaults to "item".
        TileMatrixSetId (str, optional): The TileMatrixSetId to be used in the file path. Defaults to "WebMercatorQuad".
    """
    super().__init__(endpoint, name, TileMatrixSetId)

TitilerCMREndpoint

This class contains the methods for the TiTiler CMR endpoint.

Source code in leafmap/stac.py

class TitilerCMREndpoint:
    """This class contains the methods for the TiTiler CMR endpoint."""

    def __init__(
        self,
        endpoint: Optional[str] = None,
        TileMatrixSetId: Optional[str] = "WebMercatorQuad",
    ):
        """Initialize the TitilerCMREndpoint object.

        Args:
            endpoint (str, optional): The TiTiler CMR endpoint URL.
                Defaults to "https://staging.openveda.cloud/api/titiler-cmr".
            TileMatrixSetId (str, optional): The TileMatrixSetId. Defaults to "WebMercatorQuad".
        """
        if endpoint is None:
            endpoint = os.environ.get(
                "TITILER_CMR_ENDPOINT",
                "https://staging.openveda.cloud/api/titiler-cmr",
            )
        self.endpoint = endpoint
        self.TileMatrixSetId = TileMatrixSetId

    def url_for_tilejson(self):
        """Get the URL for the TileJSON endpoint.

        Returns:
            str: The TileJSON endpoint URL.
        """
        return f"{self.endpoint}/{self.TileMatrixSetId}/tilejson.json"

    def url_for_timeseries_tilejson(self):
        """Get the URL for the time series TileJSON endpoint.

        Returns:
            str: The time series TileJSON endpoint URL.
        """
        return f"{self.endpoint}/timeseries/{self.TileMatrixSetId}/tilejson.json"

    def url_for_timeseries_gif(self, bbox, width=512, height=512):
        """Get the URL for the time series animated GIF endpoint.

        Args:
            bbox (list | tuple): Bounding box as [minx, miny, maxx, maxy].
            width (int, optional): Width of the GIF in pixels. Defaults to 512.
            height (int, optional): Height of the GIF in pixels. Defaults to 512.

        Returns:
            str: The time series GIF endpoint URL.
        """
        minx, miny, maxx, maxy = bbox
        return f"{self.endpoint}/timeseries/bbox/{minx},{miny},{maxx},{maxy}/{width}x{height}.gif"

    def url_for_statistics(self):
        """Get the URL for the statistics endpoint.

        Returns:
            str: The statistics endpoint URL.
        """
        return f"{self.endpoint}/statistics"

    def url_for_timeseries_statistics(self):
        """Get the URL for the time series statistics endpoint.

        Returns:
            str: The time series statistics endpoint URL.
        """
        return f"{self.endpoint}/timeseries/statistics"

    def url_for_timeseries(self):
        """Get the URL for the time series endpoint.

        Returns:
            str: The time series endpoint URL.
        """
        return f"{self.endpoint}/timeseries"

__init__(endpoint=None, TileMatrixSetId='WebMercatorQuad')

Initialize the TitilerCMREndpoint object.

Parameters:

Name	Type	Description	Default
endpoint	str	The TiTiler CMR endpoint URL. Defaults to "https://staging.openveda.cloud/api/titiler-cmr".	None
TileMatrixSetId	str	The TileMatrixSetId. Defaults to "WebMercatorQuad".	'WebMercatorQuad'

Source code in leafmap/stac.py

def __init__(
    self,
    endpoint: Optional[str] = None,
    TileMatrixSetId: Optional[str] = "WebMercatorQuad",
):
    """Initialize the TitilerCMREndpoint object.

    Args:
        endpoint (str, optional): The TiTiler CMR endpoint URL.
            Defaults to "https://staging.openveda.cloud/api/titiler-cmr".
        TileMatrixSetId (str, optional): The TileMatrixSetId. Defaults to "WebMercatorQuad".
    """
    if endpoint is None:
        endpoint = os.environ.get(
            "TITILER_CMR_ENDPOINT",
            "https://staging.openveda.cloud/api/titiler-cmr",
        )
    self.endpoint = endpoint
    self.TileMatrixSetId = TileMatrixSetId

url_for_statistics()

Get the URL for the statistics endpoint.

Returns:

Name	Type	Description
str		The statistics endpoint URL.

Source code in leafmap/stac.py

def url_for_statistics(self):
    """Get the URL for the statistics endpoint.

    Returns:
        str: The statistics endpoint URL.
    """
    return f"{self.endpoint}/statistics"

url_for_tilejson()

Get the URL for the TileJSON endpoint.

Returns:

Name	Type	Description
str		The TileJSON endpoint URL.

Source code in leafmap/stac.py

def url_for_tilejson(self):
    """Get the URL for the TileJSON endpoint.

    Returns:
        str: The TileJSON endpoint URL.
    """
    return f"{self.endpoint}/{self.TileMatrixSetId}/tilejson.json"

url_for_timeseries()

Get the URL for the time series endpoint.

Returns:

Name	Type	Description
str		The time series endpoint URL.

Source code in leafmap/stac.py

def url_for_timeseries(self):
    """Get the URL for the time series endpoint.

    Returns:
        str: The time series endpoint URL.
    """
    return f"{self.endpoint}/timeseries"

url_for_timeseries_gif(bbox, width=512, height=512)

Get the URL for the time series animated GIF endpoint.

Parameters:

Name	Type	Description	Default
bbox	list | tuple	Bounding box as [minx, miny, maxx, maxy].	required
width	int	Width of the GIF in pixels. Defaults to 512.	512
height	int	Height of the GIF in pixels. Defaults to 512.	512

Returns:

Name	Type	Description
str		The time series GIF endpoint URL.

Source code in leafmap/stac.py

def url_for_timeseries_gif(self, bbox, width=512, height=512):
    """Get the URL for the time series animated GIF endpoint.

    Args:
        bbox (list | tuple): Bounding box as [minx, miny, maxx, maxy].
        width (int, optional): Width of the GIF in pixels. Defaults to 512.
        height (int, optional): Height of the GIF in pixels. Defaults to 512.

    Returns:
        str: The time series GIF endpoint URL.
    """
    minx, miny, maxx, maxy = bbox
    return f"{self.endpoint}/timeseries/bbox/{minx},{miny},{maxx},{maxy}/{width}x{height}.gif"

url_for_timeseries_statistics()

Get the URL for the time series statistics endpoint.

Returns:

Name	Type	Description
str		The time series statistics endpoint URL.

Source code in leafmap/stac.py

def url_for_timeseries_statistics(self):
    """Get the URL for the time series statistics endpoint.

    Returns:
        str: The time series statistics endpoint URL.
    """
    return f"{self.endpoint}/timeseries/statistics"

url_for_timeseries_tilejson()

Get the URL for the time series TileJSON endpoint.

Returns:

Name	Type	Description
str		The time series TileJSON endpoint URL.

Source code in leafmap/stac.py

def url_for_timeseries_tilejson(self):
    """Get the URL for the time series TileJSON endpoint.

    Returns:
        str: The time series TileJSON endpoint URL.
    """
    return f"{self.endpoint}/timeseries/{self.TileMatrixSetId}/tilejson.json"

TitilerEndpoint

This class contains the methods for the titiler endpoint.

Source code in leafmap/stac.py

class TitilerEndpoint:
    """This class contains the methods for the titiler endpoint."""

    def __init__(
        self,
        endpoint: Optional[str] = None,
        name: Optional[str] = "stac",
        TileMatrixSetId: Optional[str] = "WebMercatorQuad",
    ):
        """Initialize the TiTilerEndpoint object.

        Args:
            endpoint (str, optional): The endpoint of the titiler server. Defaults to "https://giswqs-titiler-endpoint.hf.space".
            name (str, optional): The name to be used in the file path. Defaults to "stac".
            TileMatrixSetId (str, optional): The TileMatrixSetId to be used in the file path. Defaults to "WebMercatorQuad".
        """
        self.endpoint = endpoint
        self.name = name
        self.TileMatrixSetId = TileMatrixSetId

    def url_for_stac_item(self):
        return f"{self.endpoint}/{self.name}/{self.TileMatrixSetId}/tilejson.json"

    def url_for_stac_assets(self):
        return f"{self.endpoint}/{self.name}/assets"

    def url_for_stac_bounds(self):
        return f"{self.endpoint}/{self.name}/info.geojson"

    def url_for_stac_info(self):
        return f"{self.endpoint}/{self.name}/info"

    def url_for_stac_info_geojson(self):
        return f"{self.endpoint}/{self.name}/info.geojson"

    def url_for_stac_statistics(self):
        return f"{self.endpoint}/{self.name}/statistics"

    def url_for_stac_pixel_value(self, lon, lat):
        return f"{self.endpoint}/{self.name}/point/{lon},{lat}"

    def url_for_stac_wmts(self):
        return (
            f"{self.endpoint}/{self.name}/{self.TileMatrixSetId}/WMTSCapabilities.xml"
        )

__init__(endpoint=None, name='stac', TileMatrixSetId='WebMercatorQuad')

Initialize the TiTilerEndpoint object.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint of the titiler server. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
name	str	The name to be used in the file path. Defaults to "stac".	'stac'
TileMatrixSetId	str	The TileMatrixSetId to be used in the file path. Defaults to "WebMercatorQuad".	'WebMercatorQuad'

Source code in leafmap/stac.py

def __init__(
    self,
    endpoint: Optional[str] = None,
    name: Optional[str] = "stac",
    TileMatrixSetId: Optional[str] = "WebMercatorQuad",
):
    """Initialize the TiTilerEndpoint object.

    Args:
        endpoint (str, optional): The endpoint of the titiler server. Defaults to "https://giswqs-titiler-endpoint.hf.space".
        name (str, optional): The name to be used in the file path. Defaults to "stac".
        TileMatrixSetId (str, optional): The TileMatrixSetId to be used in the file path. Defaults to "WebMercatorQuad".
    """
    self.endpoint = endpoint
    self.name = name
    self.TileMatrixSetId = TileMatrixSetId

check_titiler_cmr_endpoint(titiler_cmr_endpoint=None)

Returns the TiTiler CMR endpoint.

Parameters:

Name	Type	Description	Default
titiler_cmr_endpoint	str	The TiTiler CMR endpoint URL. Defaults to None.	None

Returns:

Name	Type	Description
TitilerCMREndpoint	TitilerCMREndpoint	The TiTiler CMR endpoint object.

Source code in leafmap/stac.py

def check_titiler_cmr_endpoint(
    titiler_cmr_endpoint: Optional[str] = None,
) -> TitilerCMREndpoint:
    """Returns the TiTiler CMR endpoint.

    Args:
        titiler_cmr_endpoint (str, optional): The TiTiler CMR endpoint URL. Defaults to None.

    Returns:
        TitilerCMREndpoint: The TiTiler CMR endpoint object.
    """
    if titiler_cmr_endpoint is None:
        titiler_cmr_endpoint = os.environ.get(
            "TITILER_CMR_ENDPOINT",
            "https://staging.openveda.cloud/api/titiler-cmr",
        )

    if isinstance(titiler_cmr_endpoint, str):
        return TitilerCMREndpoint(endpoint=titiler_cmr_endpoint)
    elif isinstance(titiler_cmr_endpoint, TitilerCMREndpoint):
        return titiler_cmr_endpoint
    else:
        return TitilerCMREndpoint()

check_titiler_endpoint(titiler_endpoint=None)

Returns the default titiler endpoint.

Returns:

Type	Description
Any	The titiler endpoint.

Source code in leafmap/stac.py

def check_titiler_endpoint(titiler_endpoint: Optional[str] = None) -> Any:
    """Returns the default titiler endpoint.

    Returns:
        The titiler endpoint.
    """
    if (
        titiler_endpoint is not None
        and isinstance(titiler_endpoint, str)
        and titiler_endpoint.lower() == "local"
    ):
        titiler_endpoint = run_titiler(show_logs=False)
    elif titiler_endpoint is None:
        if os.environ.get("TITILER_ENDPOINT") is not None:
            titiler_endpoint = os.environ.get("TITILER_ENDPOINT")

            if titiler_endpoint == "planetary-computer":
                titiler_endpoint = PlanetaryComputerEndpoint()
        else:
            titiler_endpoint = "https://giswqs-titiler-endpoint.hf.space"
    elif isinstance(titiler_endpoint, str) and titiler_endpoint in [
        "planetary-computer",
        "pc",
    ]:
        titiler_endpoint = PlanetaryComputerEndpoint()
    # If it's already an endpoint object, just return it as-is

    return titiler_endpoint

cmr_animated_gif(concept_id, datetime, bbox, width=512, height=512, step='P1D', temporal_mode='point', backend='rasterio', variable=None, bands=None, bands_regex=None, expression=None, rescale=None, colormap_name=None, color_formula=None, fps=1, output=None, titiler_cmr_endpoint=None, **kwargs)

Generate animated GIF for time series from TiTiler CMR.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime range (e.g., '2024-01-01/2024-01-31').	required
bbox	list | tuple	Bounding box as [minx, miny, maxx, maxy].	required
width	int	Width of the GIF in pixels. Defaults to 512.	512
height	int	Height of the GIF in pixels. Defaults to 512.	512
step	str	ISO 8601 duration for time steps (e.g., 'P1D' for 1 day, 'P1M' for 1 month). Defaults to 'P1D'.	'P1D'
temporal_mode	str	Temporal mode - 'point' or 'range'. Defaults to 'point'.	'point'
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
bands	str | list	Band name(s) for rasterio backend.	None
bands_regex	str	Regex pattern for selecting bands (e.g., 'B[0-9][]').	None
expression	str	Band math expression (e.g., '(B05-B04)/(B05+B04)').	None
rescale	str | list	Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).	None
colormap_name	str	Name of colormap (e.g., 'thermal', 'viridis').	None
color_formula	str	Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').	None
fps	int	Frames per second for the GIF. Defaults to 1.	1
output	str	Output file path. If None, returns the URL.	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters.	{}

Returns:

Name	Type	Description
str	str	The GIF URL or output file path.

Source code in leafmap/stac.py

def cmr_animated_gif(
    concept_id: str,
    datetime: str,
    bbox: Union[List[float], Tuple[float, float, float, float]],
    width: int = 512,
    height: int = 512,
    step: str = "P1D",
    temporal_mode: str = "point",
    backend: str = "rasterio",
    variable: Optional[str] = None,
    bands: Optional[Union[str, List[str]]] = None,
    bands_regex: Optional[str] = None,
    expression: Optional[str] = None,
    rescale: Optional[Union[str, List]] = None,
    colormap_name: Optional[str] = None,
    color_formula: Optional[str] = None,
    fps: int = 1,
    output: Optional[str] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> str:
    """Generate animated GIF for time series from TiTiler CMR.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str): RFC3339 datetime range (e.g., '2024-01-01/2024-01-31').
        bbox (list | tuple): Bounding box as [minx, miny, maxx, maxy].
        width (int, optional): Width of the GIF in pixels. Defaults to 512.
        height (int, optional): Height of the GIF in pixels. Defaults to 512.
        step (str, optional): ISO 8601 duration for time steps (e.g., 'P1D' for 1 day,
            'P1M' for 1 month). Defaults to 'P1D'.
        temporal_mode (str, optional): Temporal mode - 'point' or 'range'. Defaults to 'point'.
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        bands (str | list, optional): Band name(s) for rasterio backend.
        bands_regex (str, optional): Regex pattern for selecting bands (e.g., 'B[0-9][0-9]').
        expression (str, optional): Band math expression (e.g., '(B05-B04)/(B05+B04)').
        rescale (str | list, optional): Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).
        colormap_name (str, optional): Name of colormap (e.g., 'thermal', 'viridis').
        color_formula (str, optional): Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').
        fps (int, optional): Frames per second for the GIF. Defaults to 1.
        output (str, optional): Output file path. If None, returns the URL.
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters.

    Returns:
        str: The GIF URL or output file path.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    endpoint = check_titiler_cmr_endpoint(titiler_cmr_endpoint)

    params = {
        "concept_id": concept_id,
        "datetime": datetime,
        "step": step,
        "temporal_mode": temporal_mode,
        "backend": backend,
        "fps": fps,
    }

    if variable is not None:
        params["variable"] = variable

    if bands is not None:
        if isinstance(bands, list):
            # Pass bands as a list so requests encodes as bands=B04&bands=B03&bands=B02
            params["bands"] = bands
        else:
            params["bands"] = [bands]
    elif expression is not None:
        # Extract band names from expression (e.g., "(B05-B04)/(B05+B04)" -> ["B05", "B04"])
        # Band names typically match patterns like B01, B02, B8A, etc.
        band_pattern = r"\b(B[0-9][0-9A-Za-z]?)\b"
        extracted_bands = list(dict.fromkeys(re.findall(band_pattern, expression)))
        if extracted_bands:
            params["bands"] = extracted_bands

    if bands_regex is not None:
        params["bands_regex"] = bands_regex

    if expression is not None:
        params["expression"] = expression

    if rescale is not None:
        if isinstance(rescale, list):
            if isinstance(rescale[0], list):
                params["rescale"] = ",".join([f"{r[0]},{r[1]}" for r in rescale])
            else:
                params["rescale"] = ",".join([str(r) for r in rescale])
        else:
            params["rescale"] = rescale

    if colormap_name is not None:
        params["colormap_name"] = colormap_name

    if color_formula is not None:
        params["color_formula"] = color_formula

    params.update(kwargs)

    url = endpoint.url_for_timeseries_gif(bbox, width, height)

    if output is not None:
        try:
            response = requests.get(url, params=params, timeout=120)
            response.raise_for_status()
            with open(output, "wb") as f:
                f.write(response.content)
            return output
        except Exception as e:
            print(f"Error downloading GIF from TiTiler CMR: {e}")
            return None
    else:
        # Return the full URL with parameters
        from urllib.parse import urlencode

        return f"{url}?{urlencode(params)}"

cmr_bounds(concept_id, datetime=None, backend='rasterio', variable=None, titiler_cmr_endpoint=None, **kwargs)

Get bounding box from TiTiler CMR TileJSON.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').	None
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters.	{}

Returns:

Name	Type	Description
list	List[float]	Bounding box as [minx, miny, maxx, maxy].

Source code in leafmap/stac.py

def cmr_bounds(
    concept_id: str,
    datetime: Optional[str] = None,
    backend: str = "rasterio",
    variable: Optional[str] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> List[float]:
    """Get bounding box from TiTiler CMR TileJSON.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str, optional): RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters.

    Returns:
        list: Bounding box as [minx, miny, maxx, maxy].
    """
    tilejson = cmr_tilejson(
        concept_id=concept_id,
        datetime=datetime,
        backend=backend,
        variable=variable,
        titiler_cmr_endpoint=titiler_cmr_endpoint,
        **kwargs,
    )

    if tilejson is None or "bounds" not in tilejson:
        return None

    return tilejson["bounds"]

cmr_center(concept_id, datetime=None, backend='rasterio', variable=None, titiler_cmr_endpoint=None, **kwargs)

Get center coordinates from TiTiler CMR TileJSON.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').	None
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters.	{}

Returns:

Name	Type	Description
tuple	Tuple[float, float]	Center coordinates as (longitude, latitude).

Source code in leafmap/stac.py

def cmr_center(
    concept_id: str,
    datetime: Optional[str] = None,
    backend: str = "rasterio",
    variable: Optional[str] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> Tuple[float, float]:
    """Get center coordinates from TiTiler CMR TileJSON.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str, optional): RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters.

    Returns:
        tuple: Center coordinates as (longitude, latitude).
    """
    tilejson = cmr_tilejson(
        concept_id=concept_id,
        datetime=datetime,
        backend=backend,
        variable=variable,
        titiler_cmr_endpoint=titiler_cmr_endpoint,
        **kwargs,
    )

    if tilejson is None or "center" not in tilejson:
        bounds = cmr_bounds(
            concept_id=concept_id,
            datetime=datetime,
            backend=backend,
            variable=variable,
            titiler_cmr_endpoint=titiler_cmr_endpoint,
            **kwargs,
        )
        if bounds is not None:
            return ((bounds[0] + bounds[2]) / 2, (bounds[1] + bounds[3]) / 2)
        return (None, None)

    center = tilejson["center"]
    return (center[0], center[1])

cmr_statistics(concept_id, datetime=None, geojson=None, backend='rasterio', variable=None, bands=None, titiler_cmr_endpoint=None, **kwargs)

Calculate statistics for an AOI from TiTiler CMR.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').	None
geojson	dict | str	GeoJSON geometry or file path for the AOI.	None
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
bands	str | list	Band name(s) for rasterio backend.	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters.	{}

Returns:

Name	Type	Description
dict	Dict	Statistics for the AOI.

Source code in leafmap/stac.py

def cmr_statistics(
    concept_id: str,
    datetime: Optional[str] = None,
    geojson: Optional[Union[Dict, str]] = None,
    backend: str = "rasterio",
    variable: Optional[str] = None,
    bands: Optional[Union[str, List[str]]] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> Dict:
    """Calculate statistics for an AOI from TiTiler CMR.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str, optional): RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').
        geojson (dict | str, optional): GeoJSON geometry or file path for the AOI.
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        bands (str | list, optional): Band name(s) for rasterio backend.
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters.

    Returns:
        dict: Statistics for the AOI.
    """
    import json

    if os.environ.get("USE_MKDOCS") is not None:
        return None

    endpoint = check_titiler_cmr_endpoint(titiler_cmr_endpoint)

    params = {"concept_id": concept_id, "backend": backend}

    if datetime is not None:
        params["datetime"] = datetime

    if variable is not None:
        params["variable"] = variable

    if bands is not None:
        if isinstance(bands, list):
            # Pass bands as a list so requests encodes as bands=B04&bands=B03&bands=B02
            params["bands"] = bands
        else:
            params["bands"] = [bands]

    params.update(kwargs)

    # Handle GeoJSON input
    if geojson is not None:
        if isinstance(geojson, str):
            if os.path.exists(geojson):
                with open(geojson, "r") as f:
                    geojson = json.load(f)
            else:
                geojson = json.loads(geojson)

        try:
            response = requests.post(
                endpoint.url_for_statistics(),
                params=params,
                json=geojson,
                timeout=60,
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"Error fetching statistics from TiTiler CMR: {e}")
            return None
    else:
        try:
            response = requests.get(
                endpoint.url_for_statistics(), params=params, timeout=60
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"Error fetching statistics from TiTiler CMR: {e}")
            return None

cmr_tile(concept_id, datetime=None, backend='rasterio', variable=None, bands=None, bands_regex=None, expression=None, rescale=None, colormap_name=None, color_formula=None, titiler_cmr_endpoint=None, **kwargs)

Get tile URL from TiTiler CMR.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').	None
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
bands	str | list	Band name(s) for rasterio backend.	None
bands_regex	str	Regex pattern for selecting bands (e.g., 'B[0-9][]').	None
expression	str	Band math expression (e.g., '(B05-B04)/(B05+B04)').	None
rescale	str | list	Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).	None
colormap_name	str	Name of colormap (e.g., 'thermal', 'viridis').	None
color_formula	str	Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters to pass to the TiTiler CMR endpoint.	{}

Returns:

Name	Type	Description
str	str	The tile URL template.

Source code in leafmap/stac.py

def cmr_tile(
    concept_id: str,
    datetime: Optional[str] = None,
    backend: str = "rasterio",
    variable: Optional[str] = None,
    bands: Optional[Union[str, List[str]]] = None,
    bands_regex: Optional[str] = None,
    expression: Optional[str] = None,
    rescale: Optional[Union[str, List]] = None,
    colormap_name: Optional[str] = None,
    color_formula: Optional[str] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> str:
    """Get tile URL from TiTiler CMR.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str, optional): RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        bands (str | list, optional): Band name(s) for rasterio backend.
        bands_regex (str, optional): Regex pattern for selecting bands (e.g., 'B[0-9][0-9]').
        expression (str, optional): Band math expression (e.g., '(B05-B04)/(B05+B04)').
        rescale (str | list, optional): Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).
        colormap_name (str, optional): Name of colormap (e.g., 'thermal', 'viridis').
        color_formula (str, optional): Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters to pass to the TiTiler CMR endpoint.

    Returns:
        str: The tile URL template.
    """
    tilejson = cmr_tilejson(
        concept_id=concept_id,
        datetime=datetime,
        backend=backend,
        variable=variable,
        bands=bands,
        bands_regex=bands_regex,
        expression=expression,
        rescale=rescale,
        colormap_name=colormap_name,
        color_formula=color_formula,
        titiler_cmr_endpoint=titiler_cmr_endpoint,
        **kwargs,
    )

    if tilejson is None or "tiles" not in tilejson:
        return None

    return tilejson["tiles"][0]

cmr_tilejson(concept_id, datetime=None, backend='rasterio', variable=None, bands=None, bands_regex=None, expression=None, rescale=None, colormap_name=None, color_formula=None, titiler_cmr_endpoint=None, **kwargs)

Fetch TileJSON metadata from TiTiler CMR.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').	None
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
bands	str | list	Band name(s) for rasterio backend.	None
bands_regex	str	Regex pattern for selecting bands (e.g., 'B[0-9][]').	None
expression	str	Band math expression (e.g., '(B05-B04)/(B05+B04)').	None
rescale	str | list	Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).	None
colormap_name	str	Name of colormap (e.g., 'thermal', 'viridis').	None
color_formula	str	Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters to pass to the TiTiler CMR endpoint.	{}

Returns:

Name	Type	Description
dict	Dict	TileJSON metadata dict with tiles, bounds, center, minzoom, maxzoom.

Source code in leafmap/stac.py

def cmr_tilejson(
    concept_id: str,
    datetime: Optional[str] = None,
    backend: str = "rasterio",
    variable: Optional[str] = None,
    bands: Optional[Union[str, List[str]]] = None,
    bands_regex: Optional[str] = None,
    expression: Optional[str] = None,
    rescale: Optional[Union[str, List]] = None,
    colormap_name: Optional[str] = None,
    color_formula: Optional[str] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> Dict:
    """Fetch TileJSON metadata from TiTiler CMR.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str, optional): RFC3339 datetime or range (e.g., '2024-01-15' or '2024-01-01/2024-01-31').
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        bands (str | list, optional): Band name(s) for rasterio backend.
        bands_regex (str, optional): Regex pattern for selecting bands (e.g., 'B[0-9][0-9]').
        expression (str, optional): Band math expression (e.g., '(B05-B04)/(B05+B04)').
        rescale (str | list, optional): Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).
        colormap_name (str, optional): Name of colormap (e.g., 'thermal', 'viridis').
        color_formula (str, optional): Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters to pass to the TiTiler CMR endpoint.

    Returns:
        dict: TileJSON metadata dict with tiles, bounds, center, minzoom, maxzoom.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    endpoint = check_titiler_cmr_endpoint(titiler_cmr_endpoint)

    params = {"concept_id": concept_id, "backend": backend}

    if datetime is not None:
        params["datetime"] = datetime

    if variable is not None:
        params["variable"] = variable

    if bands is not None:
        if isinstance(bands, list):
            # Pass bands as a list so requests encodes as bands=B04&bands=B03&bands=B02
            params["bands"] = bands
        else:
            params["bands"] = [bands]
    elif expression is not None:
        # Extract band names from expression (e.g., "(B05-B04)/(B05+B04)" -> ["B05", "B04"])
        # Band names typically match patterns like B01, B02, B8A, etc.
        band_pattern = r"\b(B[0-9][0-9A-Za-z]?)\b"
        extracted_bands = list(dict.fromkeys(re.findall(band_pattern, expression)))
        if extracted_bands:
            params["bands"] = extracted_bands

    if bands_regex is not None:
        params["bands_regex"] = bands_regex

    if expression is not None:
        params["expression"] = expression

    if rescale is not None:
        if isinstance(rescale, list):
            if isinstance(rescale[0], list):
                params["rescale"] = ",".join([f"{r[0]},{r[1]}" for r in rescale])
            else:
                params["rescale"] = ",".join([str(r) for r in rescale])
        else:
            params["rescale"] = rescale

    if colormap_name is not None:
        params["colormap_name"] = colormap_name

    if color_formula is not None:
        params["color_formula"] = color_formula

    params.update(kwargs)

    try:
        r = requests.get(endpoint.url_for_tilejson(), params=params, timeout=30).json()
        return r
    except Exception as e:
        print(f"Error fetching TileJSON from TiTiler CMR: {e}")
        return None

cmr_timeseries_tilejson(concept_id, datetime, step='P1D', temporal_mode='point', backend='rasterio', variable=None, bands=None, bands_regex=None, expression=None, rescale=None, colormap_name=None, color_formula=None, titiler_cmr_endpoint=None, **kwargs)

Get time series TileJSON dict from TiTiler CMR.

Parameters:

Name	Type	Description	Default
concept_id	str	NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').	required
datetime	str	RFC3339 datetime range (e.g., '2024-01-01/2024-01-31').	required
step	str	ISO 8601 duration for time steps (e.g., 'P1D' for 1 day, 'P1M' for 1 month, 'P2W' for 2 weeks). Defaults to 'P1D'.	'P1D'
temporal_mode	str	Temporal mode - 'point' or 'range'. Defaults to 'point'.	'point'
backend	str	Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr. Defaults to 'rasterio'.	'rasterio'
variable	str	Variable name for xarray backend datasets.	None
bands	str | list	Band name(s) for rasterio backend.	None
bands_regex	str	Regex pattern for selecting bands (e.g., 'B[0-9][]').	None
expression	str	Band math expression (e.g., '(B05-B04)/(B05+B04)').	None
rescale	str | list	Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).	None
colormap_name	str	Name of colormap (e.g., 'thermal', 'viridis').	None
color_formula	str	Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').	None
titiler_cmr_endpoint	str	TiTiler CMR endpoint URL.	None
**kwargs		Additional parameters.	{}

Returns:

Name	Type	Description
dict	Dict[str, Dict]	Dictionary mapping datetime strings to TileJSON metadata.

Source code in leafmap/stac.py

def cmr_timeseries_tilejson(
    concept_id: str,
    datetime: str,
    step: str = "P1D",
    temporal_mode: str = "point",
    backend: str = "rasterio",
    variable: Optional[str] = None,
    bands: Optional[Union[str, List[str]]] = None,
    bands_regex: Optional[str] = None,
    expression: Optional[str] = None,
    rescale: Optional[Union[str, List]] = None,
    colormap_name: Optional[str] = None,
    color_formula: Optional[str] = None,
    titiler_cmr_endpoint: Optional[str] = None,
    **kwargs,
) -> Dict[str, Dict]:
    """Get time series TileJSON dict from TiTiler CMR.

    Args:
        concept_id (str): NASA CMR collection concept ID (e.g., 'C2036881735-POCLOUD').
        datetime (str): RFC3339 datetime range (e.g., '2024-01-01/2024-01-31').
        step (str, optional): ISO 8601 duration for time steps (e.g., 'P1D' for 1 day,
            'P1M' for 1 month, 'P2W' for 2 weeks). Defaults to 'P1D'.
        temporal_mode (str, optional): Temporal mode - 'point' or 'range'. Defaults to 'point'.
        backend (str, optional): Backend to use - 'rasterio' for COGs or 'xarray' for NetCDF/Zarr.
            Defaults to 'rasterio'.
        variable (str, optional): Variable name for xarray backend datasets.
        bands (str | list, optional): Band name(s) for rasterio backend.
        bands_regex (str, optional): Regex pattern for selecting bands (e.g., 'B[0-9][0-9]').
        expression (str, optional): Band math expression (e.g., '(B05-B04)/(B05+B04)').
        rescale (str | list, optional): Min/max values for rescaling (e.g., '270,305' or [[270, 305]]).
        colormap_name (str, optional): Name of colormap (e.g., 'thermal', 'viridis').
        color_formula (str, optional): Color formula (e.g., 'Gamma RGB 3.5 Saturation 1.7').
        titiler_cmr_endpoint (str, optional): TiTiler CMR endpoint URL.
        **kwargs: Additional parameters.

    Returns:
        dict: Dictionary mapping datetime strings to TileJSON metadata.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    endpoint = check_titiler_cmr_endpoint(titiler_cmr_endpoint)

    params = {
        "concept_id": concept_id,
        "datetime": datetime,
        "step": step,
        "temporal_mode": temporal_mode,
        "backend": backend,
    }

    if variable is not None:
        params["variable"] = variable

    if bands is not None:
        if isinstance(bands, list):
            # Pass bands as a list so requests encodes as bands=B04&bands=B03&bands=B02
            params["bands"] = bands
        else:
            params["bands"] = [bands]
    elif expression is not None:
        # Extract band names from expression (e.g., "(B05-B04)/(B05+B04)" -> ["B05", "B04"])
        # Band names typically match patterns like B01, B02, B8A, etc.
        band_pattern = r"\b(B[0-9][0-9A-Za-z]?)\b"
        extracted_bands = list(dict.fromkeys(re.findall(band_pattern, expression)))
        if extracted_bands:
            params["bands"] = extracted_bands

    if bands_regex is not None:
        params["bands_regex"] = bands_regex

    if expression is not None:
        params["expression"] = expression

    if rescale is not None:
        if isinstance(rescale, list):
            if isinstance(rescale[0], list):
                params["rescale"] = ",".join([f"{r[0]},{r[1]}" for r in rescale])
            else:
                params["rescale"] = ",".join([str(r) for r in rescale])
        else:
            params["rescale"] = rescale

    if colormap_name is not None:
        params["colormap_name"] = colormap_name

    if color_formula is not None:
        params["color_formula"] = color_formula

    params.update(kwargs)

    try:
        r = requests.get(
            endpoint.url_for_timeseries_tilejson(), params=params, timeout=60
        ).json()
        return r
    except Exception as e:
        print(f"Error fetching time series TileJSON from TiTiler CMR: {e}")
        return None

cog_bands(url, titiler_endpoint=None)

Get band names of a Cloud Optimized GeoTIFF (COG).

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif	required
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None

Returns:

Type	Description
List	A list of band names.

Source code in leafmap/stac.py

def cog_bands(
    url: str,
    titiler_endpoint: Optional[str] = None,
) -> List:
    """Get band names of a Cloud Optimized GeoTIFF (COG).

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".

    Returns:
        A list of band names.
    """

    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    try:
        r = requests.get(
            f"{titiler_endpoint}/cog/info",
            params={
                "url": url,
            },
        ).json()

        bands = [b[0] for b in r["band_descriptions"]]
        return bands
    except Exception as e:
        print(e)
        return []

cog_bounds(url, titiler_endpoint=None)

Get the bounding box of a Cloud Optimized GeoTIFF (COG).

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif	required
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None

Returns:

Name	Type	Description
list	List	A list of values representing [left, bottom, right, top]

Source code in leafmap/stac.py

def cog_bounds(
    url: str,
    titiler_endpoint: Optional[str] = None,
) -> List:
    """Get the bounding box of a Cloud Optimized GeoTIFF (COG).

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".

    Returns:
        list: A list of values representing [left, bottom, right, top]
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    try:
        r = requests.get(
            f"{titiler_endpoint}/cog/info.geojson", params={"url": url}
        ).json()
        return r["bbox"]
    except Exception as e:
        print(e)
        return None

cog_center(url, titiler_endpoint=None)

Get the centroid of a Cloud Optimized GeoTIFF (COG).

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif	required
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None

Returns:

Type	Description
Tuple	A tuple representing (longitude, latitude).

Source code in leafmap/stac.py

def cog_center(
    url: str,
    titiler_endpoint: Optional[str] = None,
) -> Tuple:
    """Get the centroid of a Cloud Optimized GeoTIFF (COG).

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".

    Returns:
        A tuple representing (longitude, latitude).
    """

    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    bounds = cog_bounds(url, titiler_endpoint)
    if bounds is None:
        return (None, None)
    else:
        center = (
            (bounds[0] + bounds[2]) / 2,
            (bounds[1] + bounds[3]) / 2,
        )  # (lat, lon)
        return center

cog_info(url, titiler_endpoint=None, return_geojson=False)

Get band statistics of a Cloud Optimized GeoTIFF (COG).

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif	required
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None

Returns:

Name	Type	Description
list	List	A dictionary of band info.

Source code in leafmap/stac.py

def cog_info(
    url: str,
    titiler_endpoint: Optional[str] = None,
    return_geojson: Optional[bool] = False,
) -> List:
    """Get band statistics of a Cloud Optimized GeoTIFF (COG).

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".

    Returns:
        list: A dictionary of band info.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    info = "info"
    if return_geojson:
        info = "info.geojson"

    try:
        r = requests.get(
            f"{titiler_endpoint}/cog/{info}",
            params={
                "url": url,
            },
            timeout=10,
        ).json()
    except Exception as e:
        titiler_endpoint = "https://giswqs-titiler-endpoint.hf.space"
        r = requests.get(
            f"{titiler_endpoint}/cog/{info}",
            params={
                "url": url,
            },
            timeout=10,
        ).json()

    return r

cog_mosaic(links, titiler_endpoint=None, username='anonymous', layername=None, overwrite=False, verbose=True, **kwargs)

Creates a COG mosaic from a list of COG URLs.

Parameters:

Name	Type	Description	Default
links	list	A list containing COG HTTP URLs.	required
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
username	str	User name for the titiler endpoint. Defaults to "anonymous".	'anonymous'
layername	[type]	Layer name to use. Defaults to None.	None
overwrite	bool	Whether to overwrite the layer name if existing. Defaults to False.	False
verbose	bool	Whether to print out descriptive information. Defaults to True.	True

Raises:

Type	Description
Exception	If the COG mosaic fails to create.

Returns:

Type	Description
str	The tile URL for the COG mosaic.

Source code in leafmap/stac.py

def cog_mosaic(
    links: List,
    titiler_endpoint: Optional[str] = None,
    username: Optional[str] = "anonymous",
    layername=None,
    overwrite: Optional[bool] = False,
    verbose: Optional[bool] = True,
    **kwargs,
) -> str:
    """Creates a COG mosaic from a list of COG URLs.

    Args:
        links (list): A list containing COG HTTP URLs.
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".
        username (str, optional): User name for the titiler endpoint. Defaults to "anonymous".
        layername ([type], optional): Layer name to use. Defaults to None.
        overwrite (bool, optional): Whether to overwrite the layer name if existing. Defaults to False.
        verbose (bool, optional): Whether to print out descriptive information. Defaults to True.

    Raises:
        Exception: If the COG mosaic fails to create.

    Returns:
        The tile URL for the COG mosaic.
    """

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    if layername is None:
        layername = "layer_X"

    try:
        if verbose:
            print("Creating COG masaic ...")

        # Create token
        r = requests.post(
            f"{titiler_endpoint}/tokens/create",
            json={"username": username, "scope": ["mosaic:read", "mosaic:create"]},
        ).json()
        token = r["token"]

        # Create mosaic
        requests.post(
            f"{titiler_endpoint}/mosaicjson/create",
            json={
                "username": username,
                "layername": layername,
                "files": links,
                # "overwrite": overwrite
            },
            params={
                "access_token": token,
            },
        ).json()

        r2 = requests.get(
            f"{titiler_endpoint}/mosaicjson/{username}.{layername}/tilejson.json",
        ).json()

        tiles = r2["tiles"][0]

        # Convert 127.0.0.1 to localhost for Google Colab browser access
        if "google.colab" in sys.modules and "127.0.0.1" in tiles:
            tiles = tiles.replace("http://127.0.0.1", "https://localhost")

        return tiles

    except Exception as e:
        raise Exception(e)

cog_mosaic_from_file(filepath, skip_rows=0, titiler_endpoint=None, username='anonymous', layername=None, overwrite=False, verbose=True, **kwargs)

Creates a COG mosaic from a csv/txt file stored locally for through HTTP URL.

Parameters:

Name	Type	Description	Default
filepath	str	Local path or HTTP URL to the csv/txt file containing COG URLs.	required
skip_rows	int	The number of rows to skip in the file. Defaults to 0.	0
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
username	str	User name for the titiler endpoint. Defaults to "anonymous".	'anonymous'
layername	[type]	Layer name to use. Defaults to None.	None
overwrite	bool	Whether to overwrite the layer name if existing. Defaults to False.	False
verbose	bool	Whether to print out descriptive information. Defaults to True.	True

Returns:

Type	Description
str	The tile URL for the COG mosaic.

Source code in leafmap/stac.py

def cog_mosaic_from_file(
    filepath: str,
    skip_rows: Optional[int] = 0,
    titiler_endpoint: Optional[str] = None,
    username: Optional[str] = "anonymous",
    layername=None,
    overwrite: Optional[bool] = False,
    verbose: Optional[bool] = True,
    **kwargs,
) -> str:
    """Creates a COG mosaic from a csv/txt file stored locally for through HTTP URL.

    Args:
        filepath (str): Local path or HTTP URL to the csv/txt file containing COG URLs.
        skip_rows (int, optional): The number of rows to skip in the file. Defaults to 0.
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".
        username (str, optional): User name for the titiler endpoint. Defaults to "anonymous".
        layername ([type], optional): Layer name to use. Defaults to None.
        overwrite (bool, optional): Whether to overwrite the layer name if existing. Defaults to False.
        verbose (bool, optional): Whether to print out descriptive information. Defaults to True.

    Returns:
        The tile URL for the COG mosaic.
    """
    import urllib

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    links = []
    if filepath.startswith("http"):
        data = urllib.request.urlopen(filepath)
        for line in data:
            links.append(line.decode("utf-8").strip())

    else:
        with open(filepath) as f:
            links = [line.strip() for line in f.readlines()]

    links = links[skip_rows:]
    # print(links)
    mosaic = cog_mosaic(
        links, titiler_endpoint, username, layername, overwrite, verbose, **kwargs
    )
    return mosaic

cog_pixel_value(lon, lat, url, bidx, titiler_endpoint=None, verbose=True, **kwargs)

Get pixel value from COG.

Parameters:

Name	Type	Description	Default
lon	float	Longitude of the pixel.	required
lat	float	Latitude of the pixel.	required
url	str	HTTP URL to a COG, e.g., 'https://github.com/opengeos/data/releases/download/raster/Libya-2023-07-01.tif'	required
bidx	str	Dataset band indexes (e.g bidx=1, bidx=1&bidx=2&bidx=3). Defaults to None.	required
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None
verbose	bool	Print status messages. Defaults to True.	True

Returns:

Name	Type	Description
list	List	A dictionary of band info.

Source code in leafmap/stac.py

def cog_pixel_value(
    lon: float,
    lat: float,
    url: str,
    bidx: Optional[str],
    titiler_endpoint: Optional[str] = None,
    verbose: Optional[bool] = True,
    **kwargs,
) -> List:
    """Get pixel value from COG.

    Args:
        lon (float): Longitude of the pixel.
        lat (float): Latitude of the pixel.
        url (str): HTTP URL to a COG, e.g., 'https://github.com/opengeos/data/releases/download/raster/Libya-2023-07-01.tif'
        bidx (str, optional): Dataset band indexes (e.g bidx=1, bidx=1&bidx=2&bidx=3). Defaults to None.
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.
        verbose (bool, optional): Print status messages. Defaults to True.

    Returns:
        list: A dictionary of band info.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    kwargs["url"] = url
    if bidx is not None:
        kwargs["bidx"] = bidx

    r = requests.get(f"{titiler_endpoint}/cog/point/{lon},{lat}", params=kwargs).json()
    bands = cog_bands(url, titiler_endpoint)
    # if isinstance(titiler_endpoint, str):
    #     r = requests.get(f"{titiler_endpoint}/cog/point/{lon},{lat}", params=kwargs).json()
    # else:
    #     r = requests.get(
    #         titiler_endpoint.url_for_stac_pixel_value(lon, lat), params=kwargs
    #     ).json()

    if "detail" in r:
        if verbose:
            print(r["detail"])
        return None
    else:
        values = r["values"]
        result = dict(zip(bands, values))
        return result

cog_stats(url, titiler_endpoint=None)

Get band statistics of a Cloud Optimized GeoTIFF (COG).

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif	required
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None

Returns:

Name	Type	Description
list	List	A dictionary of band statistics.

Source code in leafmap/stac.py

def cog_stats(
    url: str,
    titiler_endpoint: Optional[str] = None,
) -> List:
    """Get band statistics of a Cloud Optimized GeoTIFF (COG).

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".

    Returns:
        list: A dictionary of band statistics.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    try:
        r = requests.get(
            f"{titiler_endpoint}/cog/statistics",
            params={
                "url": url,
            },
            timeout=10,
        ).json()
        return r
    except Exception as e:
        print(e)
        return None

cog_tile(url, bands=None, titiler_endpoint=None, **kwargs)

Get a tile layer from a Cloud Optimized GeoTIFF (COG). Source code adapted from https://developmentseed.org/titiler/examples/notebooks/Working_with_CloudOptimizedGeoTIFF_simple/

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif. Default to None	required
bands	list	List of bands to use. Defaults to None.	None
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
**kwargs	Any	Additional arguments to pass to the titiler endpoint. For more information about the available arguments, see https://developmentseed.org/titiler/endpoints/cog/#tiles. For example, to apply a rescaling to multiple bands, use something like rescale=["164,223","130,211","99,212"].	{}

Returns:

Type	Description
Tuple	The COG tile layer URL and bounds.

Source code in leafmap/stac.py

def cog_tile(
    url,
    bands: Optional[str] = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> Tuple:
    """Get a tile layer from a Cloud Optimized GeoTIFF (COG).
        Source code adapted from https://developmentseed.org/titiler/examples/notebooks/Working_with_CloudOptimizedGeoTIFF_simple/

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif. Default to None
        bands (list, optional): List of bands to use. Defaults to None.
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".
        **kwargs (Any): Additional arguments to pass to the titiler endpoint. For more information about the available arguments, see https://developmentseed.org/titiler/endpoints/cog/#tiles.
            For example, to apply a rescaling to multiple bands, use something like `rescale=["164,223","130,211","99,212"]`.

    Returns:
        The COG tile layer URL and bounds.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    import json

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    kwargs["url"] = url

    try:
        band_names = cog_bands(url, titiler_endpoint)
    except Exception as e:
        titiler_endpoint = "https://giswqs-titiler-endpoint.hf.space"
        band_names = cog_bands(url, titiler_endpoint)

    if isinstance(bands, str):
        bands = [bands]

    if bands is None and "bidx" not in kwargs:
        if len(band_names) >= 3:
            kwargs["bidx"] = [1, 2, 3]
    elif isinstance(bands, list) and "bidx" not in kwargs:
        if all(isinstance(x, int) for x in bands):
            if len(set(bands)) == 1:
                bands = bands[0]
            kwargs["bidx"] = bands
        elif all(isinstance(x, str) for x in bands):
            if len(set(bands)) == 1:
                bands = bands[0]
            kwargs["bidx"] = [band_names.index(x) + 1 for x in bands]
        else:
            raise ValueError("Bands must be a list of integers or strings.")

    if "palette" in kwargs:
        kwargs["colormap_name"] = kwargs["palette"].lower()
        del kwargs["palette"]

    if "bidx" not in kwargs:
        kwargs["bidx"] = [1]
    elif isinstance(kwargs["bidx"], int):
        kwargs["bidx"] = [kwargs["bidx"]]

    if len(kwargs["bidx"]) == 1 and ("colormap" not in kwargs):
        colormap = _get_image_colormap(url)
        if colormap is not None:
            kwargs["colormap"] = colormap

    if "rescale" not in kwargs and ("colormap" not in kwargs):
        try:
            stats = cog_stats(url, titiler_endpoint)
        except Exception as e:
            titiler_endpoint = "https://giswqs-titiler-endpoint.hf.space"
            stats = cog_stats(url, titiler_endpoint)

        if stats is not None and "message" not in stats:
            try:
                rescale = []
                for i in band_names:
                    rescale.append(
                        "{},{}".format(
                            stats[i]["percentile_2"],
                            stats[i]["percentile_98"],
                        )
                    )
                kwargs["rescale"] = rescale
            except Exception as e:
                pass

    if "colormap" in kwargs and isinstance(kwargs["colormap"], dict):
        kwargs["colormap"] = json.dumps(kwargs["colormap"])

    TileMatrixSetId = "WebMercatorQuad"
    if "TileMatrixSetId" in kwargs.keys():
        TileMatrixSetId = kwargs["TileMatrixSetId"]
        kwargs.pop("TileMatrixSetId")

    if "default_vis" in kwargs.keys() and kwargs["default_vis"]:
        kwargs = {"url": url}

    try:
        r = requests.get(
            f"{titiler_endpoint}/cog/{TileMatrixSetId}/tilejson.json",
            params=kwargs,
            timeout=10,
        ).json()
    except Exception as e:
        print(e)
        return None
    tiles = r["tiles"][0]
    if titiler_endpoint.startswith("https://") and tiles.startswith("http://"):
        tiles = tiles.replace("http://", "https://")

    # Convert 127.0.0.1 to localhost for Google Colab browser access
    if "google.colab" in sys.modules and "127.0.0.1" in tiles:
        tiles = tiles.replace("http://127.0.0.1", "https://localhost")

    return tiles

cog_tile_vmin_vmax(url, bands=None, titiler_endpoint=None, percentile=True, **kwargs)

Get a tile layer from a Cloud Optimized GeoTIFF (COG) and return the minimum and maximum values.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif	required
bands	list	List of bands to use. Defaults to None.	None
titiler_endpoint	str	TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
percentile	bool	Whether to use percentiles or not. Defaults to True.	True

Source code in leafmap/stac.py

def cog_tile_vmin_vmax(
    url: str,
    bands: Optional[List] = None,
    titiler_endpoint: Optional[str] = None,
    percentile: Optional[bool] = True,
    **kwargs,
) -> Tuple:
    """Get a tile layer from a Cloud Optimized GeoTIFF (COG) and return the minimum and maximum values.

    Args:
        url (str): HTTP URL to a COG, e.g., https://opendata.digitalglobe.com/events/mauritius-oil-spill/post-event/2020-08-12/105001001F1B5B00/105001001F1B5B00.tif
        bands (list, optional): List of bands to use. Defaults to None.
        titiler_endpoint (str, optional): TiTiler endpoint. Defaults to "https://giswqs-titiler-endpoint.hf.space".
        percentile (bool, optional): Whether to use percentiles or not. Defaults to True.
    Returns:
        tuple: Returns the minimum and maximum values.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    stats = cog_stats(url, titiler_endpoint)

    if isinstance(bands, str):
        bands = [bands]

    if bands is not None:
        stats = {s: stats[s] for s in stats if s in bands}

    if percentile:
        vmin = min([stats[s]["percentile_2"] for s in stats])
        vmax = max([stats[s]["percentile_98"] for s in stats])
    else:
        vmin = min([stats[s]["min"] for s in stats])
        vmax = max([stats[s]["max"] for s in stats])

    return vmin, vmax

create_mosaicjson(images, output)

Create a mosaicJSON file from a list of images.

Parameters:

Name	Type	Description	Default
images	str | list	A list of image URLs or a URL to a text file containing a list of image URLs.	required
output	str	The output mosaicJSON file path.	required

Source code in leafmap/stac.py

def create_mosaicjson(images, output):
    """Create a mosaicJSON file from a list of images.

    Args:
        images (str | list): A list of image URLs or a URL to a text file containing a list of image URLs.
        output (str): The output mosaicJSON file path.

    """
    try:
        from cogeo_mosaic.backends import MosaicBackend
        from cogeo_mosaic.mosaic import MosaicJSON
    except ImportError:
        raise ImportError(
            "cogeo-mosaic is required to use this function. "
            "Install with `pip install cogeo-mosaic`."
        )

    if isinstance(images, str):
        if images.startswith("http"):
            import urllib.request

            with urllib.request.urlopen(images) as f:
                file_contents = f.read().decode("utf-8")
                images = file_contents.strip().split("\n")
        elif not os.path.exists(images):
            raise FileNotFoundError(f"{images} does not exist.")

    elif not isinstance(images, list):
        raise ValueError("images must be a list or a URL.")

    mosaic = MosaicJSON.from_urls(images)
    with MosaicBackend(output, mosaic_def=mosaic) as f:
        f.write(overwrite=True)

download_data_catalogs(out_dir=None, quiet=True, overwrite=False)

Download geospatial data catalogs from https://github.com/giswqs/geospatial-data-catalogs.

Parameters:

Name	Type	Description	Default
out_dir	str	The output directory. Defaults to None.	None
quiet	bool	Whether to suppress the download progress bar. Defaults to True.	True
overwrite	bool	Whether to overwrite the existing data catalog. Defaults to False.	False

Returns:

Name	Type	Description
str	str	The path to the downloaded data catalog.

Source code in leafmap/stac.py

def download_data_catalogs(
    out_dir: Optional[str] = None,
    quiet: Optional[bool] = True,
    overwrite: Optional[bool] = False,
) -> str:
    """Download geospatial data catalogs from https://github.com/giswqs/geospatial-data-catalogs.

    Args:
        out_dir (str, optional): The output directory. Defaults to None.
        quiet (bool, optional): Whether to suppress the download progress bar. Defaults to True.
        overwrite (bool, optional): Whether to overwrite the existing data catalog. Defaults to False.

    Returns:
        str: The path to the downloaded data catalog.
    """
    import tempfile
    import zipfile

    import gdown

    if out_dir is None:
        out_dir = tempfile.gettempdir()
    elif not os.path.exists(out_dir):
        os.makedirs(out_dir)

    url = "https://github.com/giswqs/geospatial-data-catalogs/archive/refs/heads/master.zip"

    out_file = os.path.join(out_dir, "geospatial-data-catalogs.zip")
    work_dir = os.path.join(out_dir, "geospatial-data-catalogs-master")

    if os.path.exists(work_dir) and not overwrite:
        return work_dir
    else:
        gdown.download(url, out_file, quiet=quiet)
        with zipfile.ZipFile(out_file, "r") as zip_ref:
            zip_ref.extractall(out_dir)
        return work_dir

flatten_dict(my_dict, parent_key=False, sep='.')

Flattens a nested dictionary.

Parameters:

Name	Type	Description	Default
my_dict	dict	The dictionary to flatten.	required
parent_key	bool	Whether to include the parent key. Defaults to False.	False
sep	str	The separator to use. Defaults to '.'.	'.'

Returns:

Type	Description
Dict[str, Any]	The flattened dictionary.

Source code in leafmap/stac.py

def flatten_dict(
    my_dict: Dict[str, Any], parent_key: bool = False, sep: str = "."
) -> Dict[str, Any]:
    """Flattens a nested dictionary.

    Args:
        my_dict (dict): The dictionary to flatten.
        parent_key (bool, optional): Whether to include the parent key. Defaults to False.
        sep (str, optional): The separator to use. Defaults to '.'.

    Returns:
        The flattened dictionary.
    """

    flat_dict = {}
    for key, value in my_dict.items():
        if not isinstance(value, dict):
            flat_dict[key] = value
        else:
            sub_dict = flatten_dict(value)
            for sub_key, sub_value in sub_dict.items():
                if parent_key:
                    flat_dict[parent_key + sep + sub_key] = sub_value
                else:
                    flat_dict[sub_key] = sub_value

    return flat_dict

get_cog_link_from_stac_item(item_url)

Retrieve the URL of the GeoTIFF asset from a STAC Item.

Parameters:

Name	Type	Description	Default
item_url	str	The URL to a STAC Item JSON.	required

Returns:

Type	Description
str	URL of the first .tif asset, or None if not found.

Source code in leafmap/stac.py

def get_cog_link_from_stac_item(item_url: str) -> str:
    """
    Retrieve the URL of the GeoTIFF asset from a STAC Item.

    Args:
        item_url (str): The URL to a STAC Item JSON.

    Returns:
        URL of the first .tif asset, or None if not found.
    """
    try:
        response = requests.get(item_url)
        response.raise_for_status()
        item = response.json()

        # Look for any asset ending in .tif
        for asset_key, asset in item.get("assets", {}).items():
            href = asset.get("href", "")
            if href.endswith(".tif") or ".tif?" in href:
                return href

        print("No .tif asset found in item.")
        return None
    except Exception as e:
        print(f"Failed to retrieve STAC item: {e}")
        return None

maxar_all_items(collection_id, return_gdf=True, assets=['visual'], verbose=True, **kwargs)

Retrieve STAC items from Maxar's public STAC API.

Parameters:

Name	Type	Description	Default
collection_id	str	The collection ID, e.g., Kahramanmaras-turkey-earthquake-23 Use maxar_collections() to retrieve all available collection IDs.	required
return_gdf	bool	If True, return a GeoDataFrame. Defaults to True.	True
assets	list	A list of asset names to include in the GeoDataFrame. It can be "visual", "ms_analytic", "pan_analytic", "data-mask". Defaults to ['visual'].	['visual']
verbose	bool	If True, print progress. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the pystac Catalog.from_file() method.	{}

Returns:

Type	Description
Union[GeoDataFrame, List[Dict[str, Any]]]	If return_gdf is True, return a GeoDataFrame.

Source code in leafmap/stac.py

def maxar_all_items(
    collection_id: str,
    return_gdf: Optional[bool] = True,
    assets: Optional[List] = ["visual"],
    verbose: Optional[bool] = True,
    **kwargs: Any,
) -> Union["gpd.GeoDataFrame", List[Dict[str, Any]]]:
    """Retrieve STAC items from Maxar's public STAC API.

    Args:
        collection_id (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23
            Use maxar_collections() to retrieve all available collection IDs.
        return_gdf (bool, optional): If True, return a GeoDataFrame. Defaults to True.
        assets (list, optional): A list of asset names to include in the GeoDataFrame.
            It can be "visual", "ms_analytic", "pan_analytic", "data-mask". Defaults to ['visual'].
        verbose (bool, optional): If True, print progress. Defaults to True.
        **kwargs (Any): Additional keyword arguments to pass to the pystac Catalog.from_file() method.

    Returns:
        If return_gdf is True, return a GeoDataFrame.
    """

    child_ids = maxar_child_collections(collection_id, **kwargs)
    for index, child_id in enumerate(child_ids):
        if verbose:
            print(
                f"Processing ({str(index + 1).zfill(len(str(len(child_ids))))} out of {len(child_ids)}): {child_id} ..."
            )
        items = maxar_items(collection_id, child_id, return_gdf, assets, **kwargs)
        if return_gdf:
            if child_id == child_ids[0]:
                gdf = items
            else:
                gdf = pd.concat([gdf, items], ignore_index=True)
        else:
            if child_id == child_ids[0]:
                items_all = items
            else:
                items_all.extend(items)

    if return_gdf:
        return gdf
    else:
        return items_all

maxar_child_collections(collection_id, return_ids=True, **kwargs)

Get a list of Maxar child collections.

Parameters:

Name	Type	Description	Default
collection_id	str	The collection ID, e.g., Kahramanmaras-turkey-earthquake-23 Use maxar_collections() to retrieve all available collection IDs.	required
return_ids	bool	Whether to return the collection ids. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the pystac Catalog.from_file() method.	{}

Returns:

Name	Type	Description
list	List	A list of Maxar child collections.

Source code in leafmap/stac.py

def maxar_child_collections(
    collection_id: str, return_ids: Optional[bool] = True, **kwargs
) -> List:
    """Get a list of Maxar child collections.

    Args:
        collection_id (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23
            Use maxar_collections() to retrieve all available collection IDs.
        return_ids (bool, optional): Whether to return the collection ids. Defaults to True.
        **kwargs (Any): Additional keyword arguments to pass to the pystac Catalog.from_file() method.

    Returns:
        list: A list of Maxar child collections.
    """

    import tempfile

    from pystac import Catalog

    file_path = os.path.join(tempfile.gettempdir(), f"maxar-{collection_id}.txt")
    if return_ids:
        if os.path.exists(file_path):
            with open(file_path, "r") as f:
                return [line.strip() for line in f.readlines()]

    if "MAXAR_STAC_API" in os.environ:
        url = os.environ["MAXAR_STAC_API"]
    else:
        url = "https://maxar-opendata.s3.amazonaws.com/events/catalog.json"

    root_catalog = Catalog.from_file(url, **kwargs)

    collections = root_catalog.get_child(collection_id).get_collections()

    if return_ids:
        collection_ids = [collection.id for collection in collections]
        with open(file_path, "w") as f:
            f.write("\n".join(collection_ids))
        return collection_ids

    else:
        return collections

maxar_collection_url(collection, dtype='geojson', raw=True)

Retrieve the URL to a Maxar Open Data collection.

Parameters:

Name	Type	Description	Default
collection	str	The collection ID, e.g., Kahramanmaras-turkey-earthquake-23. Use maxar_collections() to retrieve all available collection IDs.	required
dtype	str	The data type. It can be 'geojson' or 'tsv'. Defaults to 'geojson'.	'geojson'
raw	bool	If True, return the raw URL. Defaults to True.	True

Returns:

Type	Description
str	The URL to the collection.

Source code in leafmap/stac.py

def maxar_collection_url(collection, dtype="geojson", raw=True) -> str:
    """Retrieve the URL to a Maxar Open Data collection.

    Args:
        collection (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23.
            Use maxar_collections() to retrieve all available collection IDs.
        dtype (str, optional): The data type. It can be 'geojson' or 'tsv'. Defaults to 'geojson'.
        raw (bool, optional): If True, return the raw URL. Defaults to True.

    Returns:
        The URL to the collection.
    """
    collections = maxar_collections()
    if collection not in collections:
        raise ValueError(
            f"Invalid collection name. Use maxar_collections() to retrieve all available collection IDs."
        )

    if dtype not in ["geojson", "tsv"]:
        raise ValueError(f"Invalid dtype. It can be 'geojson' or 'tsv'.")

    if raw:
        url = f"https://raw.githubusercontent.com/giswqs/maxar-open-data/master/datasets/{collection}.{dtype}"
    else:
        url = f"https://github.com/giswqs/maxar-open-data/blob/master/datasets/{collection}.{dtype}"
    return url

maxar_collections(return_ids=True, **kwargs)

Get a list of Maxar collections.

Parameters:

Name	Type	Description	Default
return_ids	bool	Whether to return the collection ids. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the pystac Catalog.from_file() method.	{}

Returns:

Name	Type	Description
list	List	A list of Maxar collections.

Source code in leafmap/stac.py

def maxar_collections(return_ids: Optional[bool] = True, **kwargs) -> List:
    """Get a list of Maxar collections.

    Args:
        return_ids (bool, optional): Whether to return the collection ids. Defaults to True.
        **kwargs (Any): Additional keyword arguments to pass to the pystac Catalog.from_file() method.

    Returns:
        list : A list of Maxar collections.
    """

    import tempfile

    import pandas as pd
    from pystac import Catalog

    if return_ids:
        url = "https://raw.githubusercontent.com/giswqs/maxar-open-data/master/datasets.csv"
        df = pd.read_csv(url)
        return df["dataset"].tolist()

    file_path = os.path.join(tempfile.gettempdir(), "maxar-collections.txt")
    if return_ids:
        if os.path.exists(file_path):
            with open(file_path, "r") as f:
                return [line.strip() for line in f.readlines()]

    if "MAXAR_STAC_API" in os.environ:
        url = os.environ["MAXAR_STAC_API"]
    else:
        url = "https://maxar-opendata.s3.amazonaws.com/events/catalog.json"

    root_catalog = Catalog.from_file(url, **kwargs)

    collections = root_catalog.get_collections()

    # if return_ids:
    #     collection_ids = [collection.id for collection in collections]
    #     with open(file_path, "w") as f:
    #         f.write("\n".join(collection_ids))

    #     return collection_ids
    # else:
    return collections

maxar_download(images, out_dir=None, quiet=False, proxy=None, speed=None, use_cookies=True, verify=True, id=None, fuzzy=False, resume=False, overwrite=False)

Download Mxar Open Data images.

Parameters:

Name	Type	Description	Default
images	str | images	The list of image links or a file path to a geojson or tsv containing the Maxar download links.	required
out_dir	str	The output directory. Defaults to None.	None
quiet	bool	Suppress terminal output. Default is False.	False
proxy	str	Proxy. Defaults to None.	None
speed	float	Download byte size per second (e.g., 256KB/s = 256 * 1024). Defaults to None.	None
use_cookies	bool	Flag to use cookies. Defaults to True.	True
verify	bool | str	Either a bool, in which case it controls whether the server's TLS certificate is verified, or a string, in which case it must be a path to a CA bundle to use. Default is True.. Defaults to True.	True
id	str	Google Drive's file ID. Defaults to None.	None
fuzzy	bool	Fuzzy extraction of Google Drive's file Id. Defaults to False.	False
resume	bool	Resume the download from existing tmp file if possible. Defaults to False.	False
overwrite	bool	Overwrite the file if it already exists. Defaults to False.	False

Source code in leafmap/stac.py

def maxar_download(
    images,
    out_dir=None,
    quiet=False,
    proxy=None,
    speed=None,
    use_cookies=True,
    verify=True,
    id=None,
    fuzzy=False,
    resume=False,
    overwrite=False,
):
    """Download Mxar Open Data images.

    Args:
        images (str | images): The list of image links or a file path to a geojson or tsv containing the Maxar download links.
        out_dir (str, optional): The output directory. Defaults to None.
        quiet (bool, optional): Suppress terminal output. Default is False.
        proxy (str, optional): Proxy. Defaults to None.
        speed (float, optional): Download byte size per second (e.g., 256KB/s = 256 * 1024). Defaults to None.
        use_cookies (bool, optional): Flag to use cookies. Defaults to True.
        verify (bool | str, optional): Either a bool, in which case it controls whether the server's TLS certificate is verified, or a string,
            in which case it must be a path to a CA bundle to use. Default is True.. Defaults to True.
        id (str, optional): Google Drive's file ID. Defaults to None.
        fuzzy (bool, optional): Fuzzy extraction of Google Drive's file Id. Defaults to False.
        resume (bool, optional): Resume the download from existing tmp file if possible. Defaults to False.
        overwrite (bool, optional): Overwrite the file if it already exists. Defaults to False.

    """
    import gdown

    if out_dir is None:
        out_dir = os.getcwd()

    if isinstance(images, str):
        if images.endswith(".geojson"):
            import geopandas as gpd

            data = gpd.read_file(images)
            images = data["visual"].tolist()
        elif images.endswith(".tsv"):
            import pandas as pd

            data = pd.read_csv(images, sep="\t")
            images = data["visual"].tolist()
        else:
            raise ValueError(f"Invalid file type. It can be 'geojson' or 'tsv'.")

    if not os.path.exists(out_dir):
        os.makedirs(out_dir)

    for index, image in enumerate(images):
        items = image.split("/")
        file_name = items[7] + ".tif"
        dir_name = items[-1].split("-")[0]
        if not os.path.exists(os.path.join(out_dir, dir_name)):
            os.makedirs(os.path.join(out_dir, dir_name))
        out_file = os.path.join(out_dir, dir_name, file_name)
        if os.path.exists(out_file) and (not overwrite):
            print(f"{out_file} already exists. Skipping...")
            continue
        if not quiet:
            print(
                f"Downloading {str(index + 1).zfill(len(str(len(images))))} out of {len(images)}: {dir_name}/{file_name}"
            )

        gdown.download(
            image, out_file, quiet, proxy, speed, use_cookies, verify, id, fuzzy, resume
        )

maxar_items(collection_id, child_id, return_gdf=True, assets=['visual'], **kwargs)

Retrieve STAC items from Maxar's public STAC API.

Parameters:

Name	Type	Description	Default
collection_id	str	The collection ID, e.g., Kahramanmaras-turkey-earthquake-23 Use maxar_collections() to retrieve all available collection IDs.	required
child_id	str	The child collection ID, e.g., 1050050044DE7E00 Use maxar_child_collections() to retrieve all available child collection IDs.	required
return_gdf	bool	If True, return a GeoDataFrame. Defaults to True.	True
assets	list	A list of asset names to include in the GeoDataFrame. It can be "visual", "ms_analytic", "pan_analytic", "data-mask". Defaults to ['visual'].	['visual']
**kwargs	Any	Additional keyword arguments to pass to the pystac Catalog.from_file() method.	{}

Returns:

Type	Description
Union[GeoDataFrame, List[Dict[str, Any]]]	If return_gdf is True, return a GeoDataFrame.

Source code in leafmap/stac.py

def maxar_items(
    collection_id: str,
    child_id: str,
    return_gdf: Optional[bool] = True,
    assets: Optional[List] = ["visual"],
    **kwargs: Any,
) -> Union["gpd.GeoDataFrame", List[Dict[str, Any]]]:
    """Retrieve STAC items from Maxar's public STAC API.

    Args:
        collection_id (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23
            Use maxar_collections() to retrieve all available collection IDs.
        child_id (str): The child collection ID, e.g., 1050050044DE7E00
            Use maxar_child_collections() to retrieve all available child collection IDs.
        return_gdf (bool, optional): If True, return a GeoDataFrame. Defaults to True.
        assets (list, optional): A list of asset names to include in the GeoDataFrame.
            It can be "visual", "ms_analytic", "pan_analytic", "data-mask". Defaults to ['visual'].
        **kwargs (Any): Additional keyword arguments to pass to the pystac Catalog.from_file() method.

    Returns:
        If return_gdf is True, return a GeoDataFrame.
    """

    import pickle
    import tempfile

    from pystac import Catalog, ItemCollection

    file_path = os.path.join(
        tempfile.gettempdir(), f"maxar-{collection_id}-{child_id}.pkl"
    )

    if os.path.exists(file_path):
        with open(file_path, "rb") as f:
            items = pickle.load(f)
        if return_gdf:
            import geopandas as gpd

            gdf = gpd.GeoDataFrame.from_features(
                pystac.ItemCollection(items).to_dict(), crs="EPSG:4326"
            )
            # convert bbox column type from list to string
            gdf["proj:bbox"] = [",".join(map(str, l)) for l in gdf["proj:bbox"]]
            if assets is not None:
                if isinstance(assets, str):
                    assets = [assets]
                elif not isinstance(assets, list):
                    raise ValueError("assets must be a list or a string.")

                for asset in assets:
                    links = []
                    for item in items:
                        if asset in item.get_assets():
                            link = item.get_assets()[asset].get_absolute_href()
                            links.append(link)
                        else:
                            links.append("")

                    gdf[asset] = links

            return gdf
        else:
            return items

    if "MAXAR_STAC_API" in os.environ:
        url = os.environ["MAXAR_STAC_API"]
    else:
        url = "https://maxar-opendata.s3.amazonaws.com/events/catalog.json"

    root_catalog = Catalog.from_file(url, **kwargs)

    collection = root_catalog.get_child(collection_id)
    child = collection.get_child(child_id)

    items = ItemCollection(child.get_all_items())

    with open(file_path, "wb") as f:
        pickle.dump(items, f)

    if return_gdf:
        import geopandas as gpd

        gdf = gpd.GeoDataFrame.from_features(
            pystac.ItemCollection(items).to_dict(), crs="EPSG:4326"
        )
        # convert bbox column type from list to string
        gdf["proj:bbox"] = [",".join(map(str, l)) for l in gdf["proj:bbox"]]
        if assets is not None:
            if isinstance(assets, str):
                assets = [assets]
            elif not isinstance(assets, list):
                raise ValueError("assets must be a list or a string.")

            for asset in assets:
                links = []
                for item in items:
                    if asset in item.get_assets():
                        link = item.get_assets()[asset].get_absolute_href()
                        links.append(link)
                    else:
                        links.append("")

                gdf[asset] = links

        return gdf
    else:
        return items

maxar_refresh()

Refresh the cached Maxar STAC items.

Source code in leafmap/stac.py

def maxar_refresh():
    """Refresh the cached Maxar STAC items."""
    import tempfile

    temp_dir = tempfile.gettempdir()
    for f in os.listdir(temp_dir):
        if f.startswith("maxar-"):
            os.remove(os.path.join(temp_dir, f))

    print("Maxar STAC items cache has been refreshed.")

maxar_search(collection, start_date=None, end_date=None, bbox=None, within=False, align=True)

Search Maxar Open Data by collection ID, date range, and/or bounding box.

Parameters:

Name	Type	Description	Default
collection	str	The collection ID, e.g., Kahramanmaras-turkey-earthquake-23. Use maxar_collections() to retrieve all available collection IDs.	required
start_date	str	The start date, e.g., 2023-01-01. Defaults to None.	None
end_date	str	The end date, e.g., 2023-12-31. Defaults to None.	None
bbox	list | GeoDataFrame	The bounding box to filter by. Can be a list of 4 coordinates or a file path or a GeoDataFrame.	None
within	bool	Whether to filter by the bounding box or the bounding box's interior. Defaults to False.	False
align	bool	If True, automatically aligns GeoSeries based on their indices. If False, the order of elements is preserved.	True

Returns:

Type	Description
GeoDataFrame	A GeoDataFrame containing the search results.

Source code in leafmap/stac.py

def maxar_search(
    collection, start_date=None, end_date=None, bbox=None, within=False, align=True
) -> "gpd.GeoDataFrame":
    """Search Maxar Open Data by collection ID, date range, and/or bounding box.

    Args:
        collection (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23.
            Use maxar_collections() to retrieve all available collection IDs.
        start_date (str, optional): The start date, e.g., 2023-01-01. Defaults to None.
        end_date (str, optional): The end date, e.g., 2023-12-31. Defaults to None.
        bbox (list | GeoDataFrame): The bounding box to filter by. Can be a list of 4 coordinates or a file path or a GeoDataFrame.
        within (bool, optional): Whether to filter by the bounding box or the bounding box's interior. Defaults to False.
        align (bool, optional): If True, automatically aligns GeoSeries based on their indices. If False, the order of elements is preserved.

    Returns:
        A GeoDataFrame containing the search results.
    """
    import datetime

    import geopandas as gpd
    import pandas as pd
    from shapely.geometry import Polygon

    collections = maxar_collections()
    if collection not in collections:
        raise ValueError(
            f"Invalid collection name. Use maxar_collections() to retrieve all available collection IDs."
        )

    url = f"https://raw.githubusercontent.com/giswqs/maxar-open-data/master/datasets/{collection}.geojson"
    data = gpd.read_file(url)

    if bbox is not None:
        bbox = gpd.GeoDataFrame(
            geometry=[Polygon.from_bounds(*bbox)],
            crs="epsg:4326",
        )
        if within:
            data = data[data.within(bbox.union_all(), align=align)]
        else:
            data = data[data.intersects(bbox.union_all(), align=align)]

    date_field = "datetime"
    new_field = f"{date_field}_temp"
    data[new_field] = pd.to_datetime(data[date_field])

    if end_date is None:
        end_date = datetime.datetime.now().strftime("%Y-%m-%d")

    if start_date is None:
        start_date = data[new_field].min()

    mask = (data[new_field] >= start_date) & (data[new_field] <= end_date)
    result = data.loc[mask]
    return result.drop(columns=[new_field], axis=1)

maxar_tile_url(collection, tile, dtype='geojson', raw=True)

Retrieve the URL to a Maxar Open Data tile.

Args:

collection (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23.
    Use maxar_collections() to retrieve all available collection IDs.
tile (str): The tile ID, e.g., 10300500D9F8E600.
dtype (str, optional): The data type. It can be 'geojson', 'json' or 'tsv'. Defaults to 'geojson'.
raw (bool, optional): If True, return the raw URL. Defaults to True.

Returns:

Type	Description
str	The URL to the tile.

Source code in leafmap/stac.py

def maxar_tile_url(collection, tile, dtype="geojson", raw=True) -> str:
    """Retrieve the URL to a Maxar Open Data tile.

    Args:

        collection (str): The collection ID, e.g., Kahramanmaras-turkey-earthquake-23.
            Use maxar_collections() to retrieve all available collection IDs.
        tile (str): The tile ID, e.g., 10300500D9F8E600.
        dtype (str, optional): The data type. It can be 'geojson', 'json' or 'tsv'. Defaults to 'geojson'.
        raw (bool, optional): If True, return the raw URL. Defaults to True.

    Returns:
        The URL to the tile.
    """

    collections = maxar_collections()
    if collection not in collections:
        raise ValueError(
            f"Invalid collection name. Use maxar_collections() to retrieve all available collection IDs."
        )

    if dtype not in ["geojson", "json", "tsv"]:
        raise ValueError(f"Invalid dtype. It can be 'geojson', 'json' or 'tsv'.")

    if raw:
        url = f"https://raw.githubusercontent.com/giswqs/maxar-open-data/master/datasets/{collection}/{tile}.{dtype}"
    else:
        url = f"https://github.com/giswqs/maxar-open-data/blob/master/datasets/{collection}/{tile}.{dtype}"

    return url

oam_search(bbox=None, start_date=None, end_date=None, limit=100, return_gdf=True, **kwargs)

Search OpenAerialMap (https://openaerialmap.org) and return a GeoDataFrame or list of image metadata.

Parameters:

Name	Type	Description	Default
bbox	list | str	The bounding box [xmin, ymin, xmax, ymax] to search within. Defaults to None.	None
start_date	str	The start date to search within, such as "2015-04-20T00:00:00.000Z". Defaults to None.	None
end_date	str	The end date to search within, such as "2015-04-21T00:00:00.000Z". Defaults to None.	None
limit	int	The maximum number of results to return. Defaults to 100.	100
return_gdf	bool	If True, return a GeoDataFrame, otherwise return a list. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the API. See https://hotosm.github.io/oam-api/	{}

Returns:

Type	Description
Union[GeoDataFrame, List[Dict[str, Any]]]	If return_gdf is True, return a GeoDataFrame. Otherwise, return a list.

Source code in leafmap/stac.py

def oam_search(
    bbox=None,
    start_date=None,
    end_date=None,
    limit: int = 100,
    return_gdf: bool = True,
    **kwargs: Any,
) -> Union["gpd.GeoDataFrame", List[Dict[str, Any]]]:
    """Search OpenAerialMap (https://openaerialmap.org) and return a GeoDataFrame or list of image metadata.

    Args:
        bbox (list | str, optional): The bounding box [xmin, ymin, xmax, ymax] to search within. Defaults to None.
        start_date (str, optional): The start date to search within, such as "2015-04-20T00:00:00.000Z". Defaults to None.
        end_date (str, optional): The end date to search within, such as "2015-04-21T00:00:00.000Z". Defaults to None.
        limit (int, optional): The maximum number of results to return. Defaults to 100.
        return_gdf (bool, optional): If True, return a GeoDataFrame, otherwise return a list. Defaults to True.
        **kwargs: Additional keyword arguments to pass to the API. See https://hotosm.github.io/oam-api/

    Returns:
        If return_gdf is True, return a GeoDataFrame. Otherwise, return a list.
    """

    import geopandas as gpd
    from shapely.geometry import Polygon

    url = "https://api.openaerialmap.org/meta"
    if bbox is not None:
        if isinstance(bbox, str):
            bbox = [float(x) for x in bbox.split(",")]
        if not isinstance(bbox, list):
            raise ValueError("bbox must be a list.")
        if len(bbox) != 4:
            raise ValueError("bbox must be a list of 4 numbers.")
        bbox = ",".join(map(str, bbox))
        kwargs["bbox"] = bbox

    if start_date is not None:
        kwargs["acquisition_from"] = start_date

    if end_date is not None:
        kwargs["acquisition_to"] = end_date

    if limit is not None:
        kwargs["limit"] = limit

    try:
        r = requests.get(url, params=kwargs).json()
        if "results" in r:
            results = []
            for result in r["results"]:
                if "geojson" in result:
                    del result["geojson"]
                if "projection" in result:
                    del result["projection"]
                if "footprint" in result:
                    del result["footprint"]
                result = flatten_dict(result)
                results.append(result)

            if not return_gdf:
                return results
            else:
                df = pd.DataFrame(results)

                polygons = [Polygon.from_bounds(*bbox) for bbox in df["bbox"]]
                gdf = gpd.GeoDataFrame(geometry=polygons, crs="epsg:4326")

                return pd.concat([gdf, df], axis=1)

        else:
            print("No results found.")
            return None

    except Exception as e:
        return None

run_titiler(show_logs=False, start_port=8000, max_port=8100, return_titiler_endpoint=False)

Run TiTiler as a background service on an available port.

This function automatically detects Google Colab and adjusts the endpoint URL to use localhost remapping (https://localhost:{port}) for compatibility.

Parameters:

Name	Type	Description	Default
show_logs	bool	If True, stream logs to the notebook output.	False
start_port	int	First port to try.	8000
max_port	int	Last port to try (exclusive).	8100
return_titiler_endpoint	bool	If True, return the titiler endpoint. Defaults to False.	False

Returns:

Name	Type	Description
tuple		(endpoint, port, process)

Source code in leafmap/stac.py

def run_titiler(
    show_logs: bool = False,
    start_port: int = 8000,
    max_port: int = 8100,
    return_titiler_endpoint: bool = False,
):
    """Run TiTiler as a background service on an available port.

    This function automatically detects Google Colab and adjusts the endpoint URL
    to use localhost remapping (https://localhost:{port}) for compatibility.

    Args:
        show_logs (bool): If True, stream logs to the notebook output.
        start_port (int): First port to try.
        max_port (int): Last port to try (exclusive).
        return_titiler_endpoint (bool): If True, return the titiler endpoint. Defaults to False.

    Returns:
        tuple: (endpoint, port, process)
    """

    import subprocess
    import socket
    import atexit
    import signal
    import time
    import threading

    def find_free_port(start, end):
        for port in range(start, end):
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                try:
                    s.bind(("127.0.0.1", port))
                    return port
                except OSError:
                    continue
        raise RuntimeError(f"No free port found between {start} and {end}")

    port = find_free_port(start_port, max_port)

    cmd = [
        "uvicorn",
        "titiler.application.main:app",
        "--host",
        "127.0.0.1",
        "--port",
        str(port),
        "--log-level",
        "info",
    ]

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE if show_logs else subprocess.DEVNULL,
        stderr=subprocess.STDOUT,
    )

    # Optionally stream logs in a background thread
    if show_logs:

        def stream_logs():
            for line in iter(proc.stdout.readline, b""):
                print(line.decode().rstrip())

        threading.Thread(target=stream_logs, daemon=True).start()

    # Wait a bit for startup
    time.sleep(2)

    # Always use http://127.0.0.1 for the endpoint so Python requests work
    # In Google Colab, tile URLs will be converted to https://localhost for browser access
    endpoint = f"http://127.0.0.1:{port}"

    if "google.colab" in sys.modules:
        print(f"🚀 TiTiler is running at {endpoint} (Google Colab mode)")
    else:
        print(f"🚀 TiTiler is running at {endpoint}")

    # Register cleanup on kernel shutdown
    def stop_titiler():
        if proc.poll() is None:
            print("🛑 Stopping TiTiler...")
            proc.send_signal(signal.SIGINT)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()

    atexit.register(stop_titiler)

    os.environ["TITILER_ENDPOINT"] = endpoint
    os.environ["TITILER_PORT"] = str(port)
    os.environ["TITILER_PROCESS"] = str(proc)

    if return_titiler_endpoint:
        return endpoint, port, proc

run_titiler_xarray(show_logs=False, start_port=8000, max_port=8100)

Start a local titiler-xarray server for Zarr/NetCDF visualization.

This function starts a local titiler-xarray server that enables dynamic tile serving from Zarr and NetCDF datasets. The server runs in the background and will be automatically stopped when the Python kernel is terminated.

Requirements

• titiler.xarray package: pip install "titiler.xarray[full]"
• uvicorn: pip install uvicorn

Parameters:

Name	Type	Description	Default
show_logs	bool	If True, stream server logs to notebook output. Defaults to False.	False
start_port	int	First port to try. Defaults to 8000.	8000
max_port	int	Last port to try (exclusive). Defaults to 8100.	8100

Returns:

Name	Type	Description
str	str	The endpoint URL of the running titiler-xarray server.

Example

endpoint = leafmap.run_titiler_xarray() m = leafmap.Map() m.add_zarr(url, variable="temperature", titiler_endpoint=endpoint)

Source code in leafmap/stac.py

def run_titiler_xarray(
    show_logs: bool = False,
    start_port: int = 8000,
    max_port: int = 8100,
) -> str:
    """Start a local titiler-xarray server for Zarr/NetCDF visualization.

    This function starts a local titiler-xarray server that enables dynamic
    tile serving from Zarr and NetCDF datasets. The server runs in the background
    and will be automatically stopped when the Python kernel is terminated.

    Requirements:
        - titiler.xarray package: pip install "titiler.xarray[full]"
        - uvicorn: pip install uvicorn

    Args:
        show_logs (bool): If True, stream server logs to notebook output.
            Defaults to False.
        start_port (int): First port to try. Defaults to 8000.
        max_port (int): Last port to try (exclusive). Defaults to 8100.

    Returns:
        str: The endpoint URL of the running titiler-xarray server.

    Example:
        >>> endpoint = leafmap.run_titiler_xarray()
        >>> m = leafmap.Map()
        >>> m.add_zarr(url, variable="temperature", titiler_endpoint=endpoint)
    """
    import subprocess
    import socket
    import atexit
    import signal
    import time
    import threading

    def find_free_port(start, end):
        for port in range(start, end):
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                try:
                    s.bind(("127.0.0.1", port))
                    return port
                except OSError:
                    continue
        raise RuntimeError(f"No free port found between {start} and {end}")

    port = find_free_port(start_port, max_port)

    # Create a simple titiler-xarray app inline
    app_code = """
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from titiler.xarray.factory import TilerFactory
from titiler.xarray.extensions import VariablesExtension

app = FastAPI(
    title="TiTiler-XArray",
    description="TiTiler application for Zarr/NetCDF datasets",
)

# Add CORS middleware to allow requests from any origin
# This is necessary for Jupyter notebooks and VSCode webviews
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

md = TilerFactory(
    router_prefix="/md",
    extensions=[VariablesExtension()],
)
app.include_router(md.router, prefix="/md", tags=["Multi Dimensional"])

@app.get("/health")
def health():
    return {"status": "ok"}
"""

    # Write the app to a temporary file
    import tempfile

    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
        f.write(app_code)
        app_file = f.name

    module_name = os.path.splitext(os.path.basename(app_file))[0]

    cmd = [
        sys.executable,
        "-m",
        "uvicorn",
        f"{module_name}:app",
        "--host",
        "127.0.0.1",
        "--port",
        str(port),
        "--log-level",
        "info",
    ]

    # Change to the temp directory so uvicorn can find the module
    cwd = os.path.dirname(app_file)

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE if show_logs else subprocess.DEVNULL,
        stderr=subprocess.STDOUT,
        cwd=cwd,
    )

    # Optionally stream logs in a background thread
    if show_logs:

        def stream_logs():
            for line in iter(proc.stdout.readline, b""):
                print(line.decode().rstrip())

        threading.Thread(target=stream_logs, daemon=True).start()

    # Wait a bit for startup
    time.sleep(3)

    # Check if the server is running
    endpoint = f"http://127.0.0.1:{port}"

    try:
        response = requests.get(f"{endpoint}/health", timeout=5)
        if response.status_code != 200:
            raise RuntimeError("Server health check failed")
    except Exception as e:
        proc.kill()
        raise RuntimeError(
            f"Failed to start titiler-xarray server: {e}\n"
            "Make sure titiler.xarray is installed: pip install 'titiler.xarray[full]'"
        )

    if "google.colab" in sys.modules:
        print(f"🚀 TiTiler-XArray is running at {endpoint} (Google Colab mode)")
    else:
        print(f"🚀 TiTiler-XArray is running at {endpoint}")

    # Register cleanup on kernel shutdown
    def stop_server():
        if proc.poll() is None:
            print("🛑 Stopping TiTiler-XArray...")
            proc.send_signal(signal.SIGINT)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
        # Clean up temp file
        try:
            os.unlink(app_file)
        except Exception as e:
            # Best-effort cleanup; ignore failures but log for debugging
            print(
                f"Warning: failed to remove temporary TiTiler-XArray app file {app_file}: {e}"
            )

    atexit.register(stop_server)

    os.environ["TITILER_XARRAY_ENDPOINT"] = endpoint

    return endpoint

stac_assets(url=None, collection=None, item=None, titiler_endpoint=None, **kwargs)

Get all assets of a STAC item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A list of assets.

Source code in leafmap/stac.py

def stac_assets(
    url: str = None,
    collection: str = None,
    item: str = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get all assets of a STAC item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A list of assets.
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    if isinstance(titiler_endpoint, str):
        r = requests.get(f"{titiler_endpoint}/stac/assets", params=kwargs).json()
    else:
        r = requests.get(titiler_endpoint.url_for_stac_assets(), params=kwargs).json()

    return r

stac_bands(url=None, collection=None, item=None, titiler_endpoint=None, **kwargs)

Get band names of a single SpatialTemporal Asset Catalog (STAC) item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A list of band names

Source code in leafmap/stac.py

def stac_bands(
    url: str = None,
    collection: str = None,
    item: str = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get band names of a single SpatialTemporal Asset Catalog (STAC) item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A list of band names
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    try:
        if isinstance(titiler_endpoint, str):
            r = requests.get(f"{titiler_endpoint}/stac/assets", params=kwargs).json()
        else:
            r = requests.get(
                titiler_endpoint.url_for_stac_assets(), params=kwargs
            ).json()

        return r
    except Exception as e:
        print(e)
        return []

stac_bounds(url=None, collection=None, item=None, titiler_endpoint=None, **kwargs)

Get the bounding box of a single SpatialTemporal Asset Catalog (STAC) item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A list of values representing [left, bottom, right, top]

Source code in leafmap/stac.py

def stac_bounds(
    url: str = None,
    collection: str = None,
    item: str = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get the bounding box of a single SpatialTemporal Asset Catalog (STAC) item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A list of values representing [left, bottom, right, top]
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
        response = requests.get(url)
        r = response.json()
        if "mosaicjson" in r:
            if "bounds" in r:
                return r["bounds"]

    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    try:
        if isinstance(titiler_endpoint, str):
            r = requests.get(
                f"{titiler_endpoint}/stac/info.geojson", params=kwargs
            ).json()
        else:
            r = requests.get(
                titiler_endpoint.url_for_stac_bounds(), params=kwargs
            ).json()

        # Try to get bounds from different response formats
        if "bbox" in r:
            bounds = r["bbox"]
        elif "properties" in r and isinstance(r["properties"], dict):
            # Handle GeoJSON Feature format (Planetary Computer)
            # Try to get bounds from properties.{asset}.bounds
            for key, value in r["properties"].items():
                if isinstance(value, dict) and "bounds" in value:
                    bounds = value["bounds"]
                    break
            else:
                # Fallback to geometry bbox if available
                if "geometry" in r and "coordinates" in r["geometry"]:
                    # Calculate bbox from geometry coordinates
                    coords = r["geometry"]["coordinates"][0]  # Assuming Polygon
                    lons = [c[0] for c in coords]
                    lats = [c[1] for c in coords]
                    bounds = [min(lons), min(lats), max(lons), max(lats)]
                else:
                    return None
        else:
            return None

        return bounds
    except Exception as e:
        print(e)
        return None

stac_center(url=None, collection=None, item=None, titiler_endpoint=None, **kwargs)

Get the centroid of a single SpatialTemporal Asset Catalog (STAC) item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
tuple	Tuple[float, float]	A tuple representing (longitude, latitude)

Source code in leafmap/stac.py

def stac_center(
    url: str = None,
    collection: str = None,
    item: str = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> Tuple[float, float]:
    """Get the centroid of a single SpatialTemporal Asset Catalog (STAC) item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        tuple: A tuple representing (longitude, latitude)
    """

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    bounds = stac_bounds(url, collection, item, titiler_endpoint, **kwargs)
    if bounds is None:
        return (None, None)
    else:
        center = (
            (bounds[0] + bounds[2]) / 2,
            (bounds[1] + bounds[3]) / 2,
        )  # (lon, lat)
        return center

stac_client(url, headers=None, parameters=None, ignore_conformance=False, modifier=None, request_modifier=None, stac_io=None, return_col_id=False, get_root=True, **kwargs)

Get the STAC client. It wraps the pystac.Client.open() method. See https://pystac-client.readthedocs.io/en/stable/api.html#pystac_client.Client.open

Parameters:

Name	Type	Description	Default
url	str	The URL of a STAC Catalog.	required
headers	dict	A dictionary of additional headers to use in all requests made to any part of this Catalog/API. Defaults to None.	None
parameters	dict	Optional dictionary of query string parameters to include in all requests. Defaults to None.	None
ignore_conformance	bool	Ignore any advertised Conformance Classes in this Catalog/API. This means that functions will skip checking conformance, and may throw an unknown error if that feature is not supported, rather than a NotImplementedError. Defaults to False.	False
modifier	function	A callable that modifies the children collection and items returned by this Client. This can be useful for injecting authentication parameters into child assets to access data from non-public sources. Defaults to None.	None
request_modifier	function	A callable that either modifies a Request instance or returns a new one. This can be useful for injecting Authentication headers and/or signing fully-formed requests (e.g. signing requests using AWS SigV4). The callable should expect a single argument, which will be an instance of requests.Request. If the callable returns a requests.Request, that will be used. Alternately, the callable may simply modify the provided request object and return None.	None
stac_io	stac_io	A StacApiIO object to use for I/O requests. Generally, leave this to the default. However in cases where customized I/O processing is required, a custom instance can be provided here.	None
return_col_id	bool	Return the collection ID. Defaults to False.	False
get_root	bool	Get the root link of the STAC object. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the pystac.Client.open() method.	{}

Returns:

Type	Description
Any	The STAC client.

Source code in leafmap/stac.py

def stac_client(
    url: str,
    headers: Optional[Dict] = None,
    parameters: Optional[Dict] = None,
    ignore_conformance: Optional[bool] = False,
    modifier: Optional[Callable] = None,
    request_modifier: Optional[Callable] = None,
    stac_io=None,
    return_col_id: Optional[bool] = False,
    get_root: Optional[bool] = True,
    **kwargs: Any,
) -> Any:
    """Get the STAC client. It wraps the pystac.Client.open() method. See
        https://pystac-client.readthedocs.io/en/stable/api.html#pystac_client.Client.open

    Args:
        url (str): The URL of a STAC Catalog.
        headers (dict, optional):  A dictionary of additional headers to use in all requests
            made to any part of this Catalog/API. Defaults to None.
        parameters (dict, optional): Optional dictionary of query string parameters to include in all requests.
            Defaults to None.
        ignore_conformance (bool, optional): Ignore any advertised Conformance Classes in this Catalog/API.
            This means that functions will skip checking conformance, and may throw an unknown error
            if that feature is not supported, rather than a NotImplementedError. Defaults to False.
        modifier (function, optional): A callable that modifies the children collection and items
            returned by this Client. This can be useful for injecting authentication parameters
            into child assets to access data from non-public sources. Defaults to None.
        request_modifier (function, optional): A callable that either modifies a Request instance or returns
            a new one. This can be useful for injecting Authentication headers and/or signing fully-formed
            requests (e.g. signing requests using AWS SigV4). The callable should expect a single argument,
            which will be an instance of requests.Request. If the callable returns a requests.Request, that
            will be used. Alternately, the callable may simply modify the provided request object and
            return None.
        stac_io (pystac.stac_io, optional): A StacApiIO object to use for I/O requests. Generally, leave
            this to the default. However in cases where customized I/O processing is required, a custom
            instance can be provided here.
        return_col_id (bool, optional): Return the collection ID. Defaults to False.
        get_root (bool, optional): Get the root link of the STAC object. Defaults to True.
        **kwargs (Any): Additional keyword arguments to pass to the pystac.Client.open() method.

    Returns:
        The STAC client.
    """
    from pystac_client import Client

    collection_id = None

    if not get_root:
        return_col_id = False

    try:
        if get_root:
            root = stac_root_link(url, return_col_id=return_col_id)
            # Handle case where root is None
            if root is None:
                if return_col_id:
                    root = (url, None)
                else:
                    root = url
        else:
            root = url

        if return_col_id:
            client = Client.open(
                root[0],
                headers,
                parameters,
                ignore_conformance,
                modifier,
                request_modifier,
                stac_io,
                **kwargs,
            )
            collection_id = root[1]
            return client, collection_id
        else:
            client = Client.open(
                root,
                headers,
                parameters,
                ignore_conformance,
                modifier,
                request_modifier,
                stac_io,
                **kwargs,
            )
            return client, client.id

    except Exception as e:
        print(e)
        return None

stac_collections(url, return_ids=False, get_root=True, **kwargs)

Get the collection IDs of a STAC catalog.

Parameters:

Name	Type	Description	Default
url	str	The STAC catalog URL.	required
return_ids	bool	Return collection IDs. Defaults to False.	False
get_root	bool	Get the root link of the STAC object. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the stac_client() method.	{}

Returns:

Type	Description
List	A list of collection IDs.

Source code in leafmap/stac.py

def stac_collections(
    url: str, return_ids: Optional[bool] = False, get_root=True, **kwargs
) -> List:
    """Get the collection IDs of a STAC catalog.

    Args:
        url (str): The STAC catalog URL.
        return_ids (bool, optional): Return collection IDs. Defaults to False.
        get_root (bool, optional): Get the root link of the STAC object. Defaults to True.
        **kwargs (Any): Additional keyword arguments to pass to the stac_client() method.

    Returns:
        A list of collection IDs.
    """
    try:
        client, _ = stac_client(url, get_root=get_root, **kwargs)
        collections = client.get_all_collections()

        if return_ids:
            return [c.id for c in collections]
        else:
            return collections

    except Exception as e:
        print(e)
        return None

stac_info(url=None, collection=None, item=None, assets=None, titiler_endpoint=None, **kwargs)

Get band info of a STAC item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
assets	str | list	The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A dictionary of band info.

Source code in leafmap/stac.py

def stac_info(
    url: str = None,
    collection: str = None,
    item: str = None,
    assets: Union[str, List] = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get band info of a STAC item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        assets (str | list): The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A dictionary of band info.
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item
    if assets is not None:
        kwargs["assets"] = assets

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    if isinstance(titiler_endpoint, str):
        r = requests.get(f"{titiler_endpoint}/stac/info", params=kwargs).json()
    else:
        r = requests.get(titiler_endpoint.url_for_stac_info(), params=kwargs).json()

    return r

stac_info_geojson(url=None, collection=None, item=None, assets=None, titiler_endpoint=None, **kwargs)

Get band info of a STAC item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
assets	str | list	The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A dictionary of band info.

Source code in leafmap/stac.py

def stac_info_geojson(
    url: str = None,
    collection: str = None,
    item: str = None,
    assets: Union[str, List] = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get band info of a STAC item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        assets (str | list): The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A dictionary of band info.
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item
    if assets is not None:
        kwargs["assets"] = assets

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    if isinstance(titiler_endpoint, str):
        r = requests.get(f"{titiler_endpoint}/stac/info.geojson", params=kwargs).json()
    else:
        r = requests.get(
            titiler_endpoint.url_for_stac_info_geojson(), params=kwargs
        ).json()

    return r

stac_min_max(url=None, collection=None, item=None, assets=None, titiler_endpoint=None, **kwargs)

Get the min and max values of a STAC item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
assets	str | list	The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A dictionary of band statistics.

Source code in leafmap/stac.py

def stac_min_max(
    url: str = None,
    collection: str = None,
    item: str = None,
    assets: Union[str, List] = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get the min and max values of a STAC item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        assets (str | list): The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A dictionary of band statistics.
    """

    stats = stac_stats(url, collection, item, assets, titiler_endpoint, **kwargs)

    values = stats.values()

    try:
        min_values = [v["min"] for v in values]
        max_values = [v["max"] for v in values]

        return min(min_values), max(max_values)
    except Exception as e:
        return None, None

stac_object_type(url, **kwargs)

Get the STAC object type.

Parameters:

Name	Type	Description	Default
url	str	The STAC object URL.	required
**kwargs	Any	Keyword arguments for pystac.STACObject.from_file(). Defaults to None.	{}

Returns:

Type	Description
str	The STAC object type, can be catalog, collection, or item.

Source code in leafmap/stac.py

def stac_object_type(url: str, **kwargs) -> str:
    """Get the STAC object type.

    Args:
        url (str): The STAC object URL.
        **kwargs (Any): Keyword arguments for pystac.STACObject.from_file(). Defaults to None.

    Returns:
        The STAC object type, can be catalog, collection, or item.
    """
    try:
        obj = pystac.STACObject.from_file(url, **kwargs)

        if isinstance(obj, pystac.Collection):
            return "collection"
        elif isinstance(obj, pystac.Item):
            return "item"
        elif isinstance(obj, pystac.Catalog):
            return "catalog"

    except Exception as e:
        print(e)
        return None

stac_pixel_value(lon, lat, url=None, collection=None, item=None, assets=None, titiler_endpoint=None, verbose=True, **kwargs)

Get pixel value from STAC assets.

Parameters:

Name	Type	Description	Default
lon	float	Longitude of the pixel.	required
lat	float	Latitude of the pixel.	required
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
assets	str | list	The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None
verbose	bool	Print out the error message. Defaults to True.	True

Returns:

Type	Description
Dict[str, Any]	A dictionary of pixel values for each asset.

Source code in leafmap/stac.py

def stac_pixel_value(
    lon: float,
    lat: float,
    url: str = None,
    collection: str = None,
    item: str = None,
    assets: Union[str, List] = None,
    titiler_endpoint: Optional[str] = None,
    verbose: Optional[bool] = True,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Get pixel value from STAC assets.

    Args:
        lon (float): Longitude of the pixel.
        lat (float): Latitude of the pixel.
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        assets (str | list): The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.
        verbose (bool, optional): Print out the error message. Defaults to True.

    Returns:
        A dictionary of pixel values for each asset.
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item

    if assets is None:
        assets = stac_assets(
            url=url,
            collection=collection,
            item=item,
            titiler_endpoint=titiler_endpoint,
        )
    kwargs["assets"] = assets

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    if isinstance(titiler_endpoint, str):
        r = requests.get(
            f"{titiler_endpoint}/stac/point/{lon},{lat}", params=kwargs
        ).json()
    else:
        r = requests.get(
            titiler_endpoint.url_for_stac_pixel_value(lon, lat), params=kwargs
        ).json()

    if "detail" in r:
        if verbose:
            print(r["detail"])
        return None
    else:
        values = r["values"]
        if isinstance(assets, str):
            assets = assets.split(",")
        result = dict(zip(assets, values))
        return result

stac_root_link(url, return_col_id=False, **kwargs)

Get the root link of a STAC object.

Parameters:

Name	Type	Description	Default
url	str	The STAC object URL.	required
return_col_id	bool	Return the collection ID if the STAC object is a collection. Defaults to False.	False
**kwargs	Any	Keyword arguments for pystac.STACObject.from_file().	{}

Returns:

Type	Description
str	The root link of the STAC object. Returns None if no root link found.
str	If return_col_id is True, returns (root_link, collection_id).

Raises:

Type	Description
ValueError	If FeatureCollection contains no features.

Source code in leafmap/stac.py

def stac_root_link(url: str, return_col_id: Optional[bool] = False, **kwargs) -> str:
    """Get the root link of a STAC object.

    Args:
        url (str): The STAC object URL.
        return_col_id (bool, optional): Return the collection ID if the STAC object
            is a collection. Defaults to False.
        **kwargs (Any): Keyword arguments for pystac.STACObject.from_file().

    Returns:
        The root link of the STAC object. Returns None if no root link found.
        If return_col_id is True, returns (root_link, collection_id).

    Raises:
        ValueError: If FeatureCollection contains no features.
    """
    collection_id = None
    try:
        response = requests.get(url, **kwargs)
        response.raise_for_status()
        data = response.json()

        # Handle FeatureCollection from /items endpoint
        if data.get("type") == "FeatureCollection":
            features = data.get("features", [])
            if not features:
                raise ValueError("FeatureCollection contains no features.")
            item = pystac.Item.from_dict(features[0])
            collection_id = item.collection_id
            root = item.get_root_link()
            return (
                (root.get_href() if root else item.get_self_href(), collection_id)
                if return_col_id
                else root.get_href() if root else item.get_self_href()
            )

        # Handle STAC API objects (Collection, Catalog, etc.)
        links = data.get("links", [])
        root_href = None
        self_href = None

        for link in links:
            if link.get("rel") == "root":
                root_href = link.get("href")
            elif link.get("rel") == "self":
                self_href = link.get("href")

        # Get collection/catalog ID if requested
        if return_col_id:
            if data.get("type") in ["Collection", "Catalog"]:
                collection_id = data.get("id")

        # Determine the root href:
        # 1. If there's an explicit root link, use it
        # 2. If there's no root link but there's a self link, the self link is the root
        #    (this is common for root catalogs in STAC APIs)
        # 3. Otherwise, fall back to parsing as a regular STAC object
        if root_href:
            return (root_href, collection_id) if return_col_id else root_href
        elif self_href and not root_href:
            # When there's no root link, self link often indicates this IS the root
            return (self_href, collection_id) if return_col_id else self_href

        # Fallback: parse as regular STAC object using pystac
        obj = pystac.STACObject.from_dict(data)
        if isinstance(obj, (pystac.Collection, pystac.Catalog)):
            collection_id = obj.id

        root = obj.get_root_link()
        href = root.get_href() if root else obj.get_self_href()
        return (href, collection_id) if return_col_id else href

    except Exception as e:
        print(f"Failed to resolve STAC root from {url}: {e}")
        return (None, None) if return_col_id else None

stac_search(url, method='POST', max_items=None, limit=100, ids=None, collections=None, bbox=None, intersects=None, datetime=None, query=None, filter=None, filter_lang=None, sortby=None, fields=None, get_collection=False, get_items=False, get_assets=False, get_links=False, get_gdf=False, get_info=False, get_root=True, **kwargs)

Search a STAC API. The function wraps the pysatc_client.Client.search() method. See https://pystac-client.readthedocs.io/en/stable/api.html#pystac_client.Client.search

Parameters:

Name	Type	Description	Default
url	str	The STAC API URL.	required
method	str	The HTTP method to use when making a request to the service. This must be either "GET", "POST", or None. If None, this will default to "POST". If a "POST" request receives a 405 status for the response, it will automatically retry with "GET" for all subsequent requests. Defaults to "POST".	'POST'
max_items	init	The maximum number of items to return from the search, even if there are more matching results. This client to limit the total number of Items returned from the items(), item_collections(), and items_as_dicts methods(). The client will continue to request pages of items until the number of max items is reached. This parameter defaults to 100. Setting this to None will allow iteration over a possibly very large number of results.. Defaults to None.	None
limit	int	A recommendation to the service as to the number of items to return per page of results. Defaults to 100.	100
ids	list	List of one or more Item ids to filter on. Defaults to None.	None
collections	list	List of one or more Collection IDs or pystac.Collection instances. Only Items in one of the provided Collections will be searched. Defaults to None.	None
bbox	list | tuple	A list, tuple, or iterator representing a bounding box of 2D or 3D coordinates. Results will be filtered to only those intersecting the bounding box. Defaults to None.	None
intersects	str | dict	A string or dictionary representing a GeoJSON geometry, or an object that implements a geo_interface property, as supported by several libraries including Shapely, ArcPy, PySAL, and geojson. Results filtered to only those intersecting the geometry. Defaults to None.	None
datetime	str	Either a single datetime or datetime range used to filter results. You may express a single datetime using a datetime.datetime instance, a RFC 3339-compliant timestamp, or a simple date string (see below). Instances of datetime.datetime may be either timezone aware or unaware. Timezone aware instances will be converted to a UTC timestamp before being passed to the endpoint. Timezone unaware instances are assumed to represent UTC timestamps. You may represent a datetime range using a "/" separated string as described in the spec, or a list, tuple, or iterator of 2 timestamps or datetime instances. For open-ended ranges, use either ".." ('2020-01-01:00:00:00Z/..', ['2020-01-01:00:00:00Z', '..']) or a value of None (['2020-01-01:00:00:00Z', None]). If using a simple date string, the datetime can be specified in YYYY-mm-dd format, optionally truncating to YYYY-mm or just YYYY. Simple date strings will be expanded to include the entire time period. Defaults to None.	None
query	list	List or JSON of query parameters as per the STAC API query extension. such as {"eo:cloud_cover":{"lt":10}}. Defaults to None.	None
filter	dict	JSON of query parameters as per the STAC API filter extension. Defaults to None.	None
filter_lang	str	Language variant used in the filter body. If filter is a dictionary or not provided, defaults to ‘cql2-json’. If filter is a string, defaults to cql2-text. Defaults to None.	None
sortby	str | list	A single field or list of fields to sort the response by. such as [{ 'field': 'properties.eo:cloud_cover', 'direction': 'asc' }]. Defaults to None.	None
fields	list	A list of fields to include in the response. Note this may result in invalid STAC objects, as they may not have required fields. Use items_as_dicts to avoid object unmarshalling errors. Defaults to None.	None
get_collection	bool	True to return a pystac.ItemCollection. Defaults to False.	False
get_items	bool	True to return a list of pystac.Item. Defaults to False.	False
get_assets	bool	True to return a list of pystac.Asset. Defaults to False.	False
get_links	bool	True to return a list of links. Defaults to False.	False
get_gdf	bool	True to return a GeoDataFrame. Defaults to False.	False
get_info	bool	True to return a dictionary of STAC items. Defaults to False.	False
get_root	bool	Get the root link of the STAC object. Defaults to True.	True
**kwargs	Any	Additional keyword arguments to pass to the stac_client() function.	{}

Returns:

Type	Description
List	The search results as a list of links or a pystac.ItemCollection.

Source code in leafmap/stac.py

def stac_search(
    url: str,
    method: Optional[str] = "POST",
    max_items: Optional[int] = None,
    limit: Optional[int] = 100,
    ids: Optional[List] = None,
    collections: Optional[List] = None,
    bbox: Optional[Union[List, Tuple]] = None,
    intersects: Optional[Union[str, Dict]] = None,
    datetime: Optional[str] = None,
    query: Optional[List] = None,
    filter: Optional[Dict] = None,
    filter_lang: Optional[str] = None,
    sortby: Optional[Union[List, str]] = None,
    fields: Optional[List] = None,
    get_collection: Optional[bool] = False,
    get_items: Optional[bool] = False,
    get_assets: Optional[bool] = False,
    get_links: Optional[bool] = False,
    get_gdf: Optional[bool] = False,
    get_info: Optional[bool] = False,
    get_root: Optional[bool] = True,
    **kwargs,
) -> List:
    """Search a STAC API. The function wraps the pysatc_client.Client.search() method. See
        https://pystac-client.readthedocs.io/en/stable/api.html#pystac_client.Client.search

    Args:
        url (str): The STAC API URL.
        method (str, optional): The HTTP method to use when making a request to the service.
            This must be either "GET", "POST", or None. If None, this will default to "POST".
            If a "POST" request receives a 405 status for the response, it will automatically
            retry with "GET" for all subsequent requests. Defaults to "POST".
        max_items (init, optional): The maximum number of items to return from the search,
            even if there are more matching results. This client to limit the total number of
            Items returned from the items(), item_collections(), and items_as_dicts methods().
            The client will continue to request pages of items until the number of max items
            is reached. This parameter defaults to 100. Setting this to None will allow iteration
            over a possibly very large number of results.. Defaults to None.
        limit (int, optional): A recommendation to the service as to the number of items to
            return per page of results. Defaults to 100.
        ids (list, optional): List of one or more Item ids to filter on. Defaults to None.
        collections (list, optional): List of one or more Collection IDs or pystac.Collection instances.
            Only Items in one of the provided Collections will be searched. Defaults to None.
        bbox (list | tuple, optional): A list, tuple, or iterator representing a bounding box of 2D
            or 3D coordinates. Results will be filtered to only those intersecting the bounding box.
            Defaults to None.
        intersects (str | dict, optional):  A string or dictionary representing a GeoJSON geometry, or
            an object that implements a __geo_interface__ property, as supported by several
            libraries including Shapely, ArcPy, PySAL, and geojson. Results filtered to only
            those intersecting the geometry. Defaults to None.
        datetime (str, optional): Either a single datetime or datetime range used to filter results.
            You may express a single datetime using a datetime.datetime instance, a RFC 3339-compliant
            timestamp, or a simple date string (see below). Instances of datetime.datetime may be either
            timezone aware or unaware. Timezone aware instances will be converted to a UTC timestamp
            before being passed to the endpoint. Timezone unaware instances are assumed to represent
            UTC timestamps. You may represent a datetime range using a "/" separated string as described
            in the spec, or a list, tuple, or iterator of 2 timestamps or datetime instances.
            For open-ended ranges, use either ".." ('2020-01-01:00:00:00Z/..', ['2020-01-01:00:00:00Z', '..'])
            or a value of None (['2020-01-01:00:00:00Z', None]). If using a simple date string,
            the datetime can be specified in YYYY-mm-dd format, optionally truncating to
            YYYY-mm or just YYYY. Simple date strings will be expanded to include the entire
            time period. Defaults to None.
        query (list, optional): List or JSON of query parameters as per the STAC API query extension.
            such as {"eo:cloud_cover":{"lt":10}}. Defaults to None.
        filter (dict, optional): JSON of query parameters as per the STAC API filter extension. Defaults to None.
        filter_lang (str, optional): Language variant used in the filter body. If filter is a dictionary
            or not provided, defaults to ‘cql2-json’. If filter is a string, defaults to cql2-text. Defaults to None.
        sortby (str | list, optional): A single field or list of fields to sort the response by.
            such as [{ 'field': 'properties.eo:cloud_cover', 'direction': 'asc' }]. Defaults to None.
        fields (list, optional): A list of fields to include in the response. Note this may result in
            invalid STAC objects, as they may not have required fields. Use items_as_dicts to avoid object
            unmarshalling errors. Defaults to None.
        get_collection (bool, optional): True to return a pystac.ItemCollection. Defaults to False.
        get_items (bool, optional): True to return a list of pystac.Item. Defaults to False.
        get_assets (bool, optional): True to return a list of pystac.Asset. Defaults to False.
        get_links (bool, optional): True to return a list of links. Defaults to False.
        get_gdf (bool, optional): True to return a GeoDataFrame. Defaults to False.
        get_info (bool, optional): True to return a dictionary of STAC items. Defaults to False.
        get_root (bool, optional): Get the root link of the STAC object. Defaults to True.
        **kwargs (Any): Additional keyword arguments to pass to the stac_client() function.

    Returns:
        The search results as a list of links or a pystac.ItemCollection.
    """

    client, collection_id = stac_client(
        url, return_col_id=True, get_root=get_root, **kwargs
    )

    if client is None:
        return None
    else:
        if isinstance(intersects, dict) and "geometry" in intersects:
            intersects = intersects["geometry"]

        if collection_id is not None and collections is None:
            collections = [collection_id]

        search = client.search(
            method=method,
            max_items=max_items,
            limit=limit,
            ids=ids,
            collections=collections,
            bbox=bbox,
            intersects=intersects,
            datetime=datetime,
            query=query,
            filter=filter,
            filter_lang=filter_lang,
            sortby=sortby,
            fields=fields,
        )

        if get_collection:
            return search.item_collection()
        elif get_items:
            return list(search.items())
        elif get_assets:
            assets = {}
            for item in search.items():
                assets[item.id] = {}
                for key, value in item.get_assets().items():
                    assets[item.id][key] = value.href
            return assets
        elif get_links:
            return [item.get_self_href() for item in search.items()]
        elif get_gdf:
            import geopandas as gpd

            gdf = gpd.GeoDataFrame.from_features(
                search.item_collection().to_dict(), crs="EPSG:4326"
            )
            return gdf
        elif get_info:
            items = search.items()
            info = {}
            for item in items:
                info[item.id] = {
                    "id": item.id,
                    "href": item.get_self_href(),
                    "bands": list(item.get_assets().keys()),
                    "assets": item.get_assets(),
                }
            return info
        else:
            return search

stac_search_to_df(search, **kwargs)

Convert STAC search result to a DataFrame.

Parameters:

Name	Type	Description	Default
search	item_search	The search result returned by leafmap.stac_search().	required
**kwargs	Any	Additional keyword arguments to pass to the DataFrame.drop() function.	{}

Returns:

Type	Description
DataFrame	A Pandas DataFrame object.

Source code in leafmap/stac.py

def stac_search_to_df(search, **kwargs) -> pd.DataFrame:
    """Convert STAC search result to a DataFrame.

    Args:
        search (pystac_client.item_search): The search result returned by leafmap.stac_search().
        **kwargs (Any): Additional keyword arguments to pass to the DataFrame.drop() function.

    Returns:
        A Pandas DataFrame object.
    """
    gdf = stac_search_to_gdf(search)
    return gdf.drop(columns=["geometry"], **kwargs)

stac_search_to_dict(search, **kwargs)

Convert STAC search result to a dictionary.

Parameters:

Name	Type	Description	Default
search	item_search	The search result returned by leafmap.stac_search().	required

Returns:

Name	Type	Description
dict	Dict	A dictionary of STAC items, with the stac item id as the key, and the stac item as the value.

Source code in leafmap/stac.py

def stac_search_to_dict(search, **kwargs) -> Dict:
    """Convert STAC search result to a dictionary.

    Args:
        search (pystac_client.item_search): The search result returned by leafmap.stac_search().

    Returns:
        dict: A dictionary of STAC items, with the stac item id as the key, and the stac item as the value.
    """

    items = list(search.item_collection())
    info = {}
    for item in items:
        info[item.id] = {
            "id": item.id,
            "href": item.get_self_href(),
            "bands": list(item.get_assets().keys()),
            "assets": item.get_assets(),
        }
        links = {}
        assets = item.get_assets()
        for key, value in assets.items():
            links[key] = value.href
        info[item.id]["links"] = links
    return info

stac_search_to_gdf(search, **kwargs)

Convert STAC search result to a GeoDataFrame.

Parameters:

Name	Type	Description	Default
search	item_search	The search result returned by leafmap.stac_search().	required
**kwargs	Any	Additional keyword arguments to pass to the GeoDataFrame.from_features() function.	{}

Returns:

Type	Description
GeoDataFrame	A GeoPandas GeoDataFrame object.

Source code in leafmap/stac.py

def stac_search_to_gdf(search, **kwargs: Any) -> "gpd.GeoDataFrame":
    """Convert STAC search result to a GeoDataFrame.

    Args:
        search (pystac_client.item_search): The search result returned by leafmap.stac_search().
        **kwargs (Any): Additional keyword arguments to pass to the GeoDataFrame.from_features() function.

    Returns:
        A GeoPandas GeoDataFrame object.
    """
    import geopandas as gpd

    gdf = gpd.GeoDataFrame.from_features(
        search.item_collection().to_dict(), crs="EPSG:4326", **kwargs
    )
    return gdf

stac_search_to_list(search, **kwargs)

Convert STAC search result to a list.

Parameters:

Name	Type	Description	Default
search	item_search	The search result returned by leafmap.stac_search().	required

Returns:

Name	Type	Description
list	List	A list of STAC items.

Source code in leafmap/stac.py

def stac_search_to_list(search, **kwargs) -> List:
    """Convert STAC search result to a list.

    Args:
        search (pystac_client.item_search): The search result returned by leafmap.stac_search().

    Returns:
        list: A list of STAC items.
    """

    return search.item_collections()

stac_stats(url=None, collection=None, item=None, assets=None, titiler_endpoint=None, **kwargs)

Get band statistics of a STAC item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
assets	str | list	The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.	None

Returns:

Name	Type	Description
list	List	A dictionary of band statistics.

Source code in leafmap/stac.py

def stac_stats(
    url: str = None,
    collection: str = None,
    item: str = None,
    assets: Union[str, List] = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> List:
    """Get band statistics of a STAC item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        assets (str | list): The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "planetary-computer", "pc". Defaults to None.

    Returns:
        list: A dictionary of band statistics.
    """

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item
    if assets is not None:
        kwargs["assets"] = assets

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    if isinstance(titiler_endpoint, str):
        r = requests.get(f"{titiler_endpoint}/stac/statistics", params=kwargs).json()
    else:
        r = requests.get(
            titiler_endpoint.url_for_stac_statistics(), params=kwargs
        ).json()

    return r

stac_tile(url=None, collection=None, item=None, assets=None, bands=None, titiler_endpoint=None, **kwargs)

Get a tile layer from a single SpatialTemporal Asset Catalog (STAC) item.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json	None
collection	str	The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.	None
item	str	The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.	None
assets	str | list	The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].	None
bands	list	A list of band names, e.g., ["SR_B7", "SR_B5", "SR_B4"]	None
titiler_endpoint	str	TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "https://planetarycomputer.microsoft.com/api/data/v1", "planetary-computer", "pc". Defaults to None.	None

Returns:

Type	Description
str	The STAC tile layer URL.

Source code in leafmap/stac.py

def stac_tile(
    url: str = None,
    collection: str = None,
    item: str = None,
    assets: Union[str, List] = None,
    bands: list = None,
    titiler_endpoint: Optional[str] = None,
    **kwargs,
) -> str:
    """Get a tile layer from a single SpatialTemporal Asset Catalog (STAC) item.

    Args:
        url (str): HTTP URL to a STAC item, e.g., https://canada-spot-ortho.s3.amazonaws.com/canada_spot_orthoimages/canada_spot5_orthoimages/S5_2007/S5_11055_6057_20070622/S5_11055_6057_20070622.json
        collection (str): The Microsoft Planetary Computer STAC collection ID, e.g., landsat-8-c2-l2.
        item (str): The Microsoft Planetary Computer STAC item ID, e.g., LC08_L2SP_047027_20201204_02_T1.
        assets (str | list): The Microsoft Planetary Computer STAC asset ID, e.g., ["SR_B7", "SR_B5", "SR_B4"].
        bands (list): A list of band names, e.g., ["SR_B7", "SR_B5", "SR_B4"]
        titiler_endpoint (str, optional): TiTiler endpoint, e.g., "https://giswqs-titiler-endpoint.hf.space", "https://planetarycomputer.microsoft.com/api/data/v1", "planetary-computer", "pc". Defaults to None.

    Returns:
        The STAC tile layer URL.
    """
    import json

    if url is None and collection is None:
        raise ValueError("Either url or collection must be specified.")

    if collection is not None and titiler_endpoint is None:
        titiler_endpoint = "planetary-computer"

    if isinstance(url, pystac.Item):
        try:
            url = url.self_href
        except Exception as e:
            print(e)

    if url is not None:
        kwargs["url"] = url
    if collection is not None:
        kwargs["collection"] = collection
    if item is not None:
        kwargs["item"] = item

    if "palette" in kwargs:
        kwargs["colormap_name"] = kwargs["palette"].lower()
        del kwargs["palette"]

    if isinstance(bands, list) and len(set(bands)) == 1:
        bands = bands[0]

    if isinstance(assets, list) and len(set(assets)) == 1:
        assets = assets[0]

    titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    if "expression" in kwargs and ("asset_as_band" not in kwargs):
        kwargs["asset_as_band"] = True

    mosaic_json = False

    if isinstance(titiler_endpoint, PlanetaryComputerEndpoint):
        if isinstance(bands, str):
            bands = bands.split(",")
        if isinstance(assets, str):
            assets = assets.split(",")
        if assets is None and (bands is not None):
            assets = bands
        else:
            kwargs["bidx"] = bands

        kwargs["assets"] = assets

        if (
            (assets is not None)
            and ("asset_expression" not in kwargs)
            and ("expression" not in kwargs)
            and ("rescale" not in kwargs)
        ):
            try:
                stats = stac_stats(
                    collection=collection,
                    item=item,
                    assets=assets,
                    titiler_endpoint=titiler_endpoint,
                )
            except Exception as e:
                fallback_endpoint = "https://giswqs-titiler-endpoint.hf.space"
                stats = stac_stats(
                    collection=collection,
                    item=item,
                    assets=assets,
                    titiler_endpoint=fallback_endpoint,
                )

            if "detail" not in stats:
                try:
                    percentile_2 = min([stats[s]["percentile_2"] for s in stats])
                    percentile_98 = max([stats[s]["percentile_98"] for s in stats])
                except:
                    percentile_2 = min(
                        [
                            stats[s][list(stats[s].keys())[0]]["percentile_2"]
                            for s in stats
                        ]
                    )
                    percentile_98 = max(
                        [
                            stats[s][list(stats[s].keys())[0]]["percentile_98"]
                            for s in stats
                        ]
                    )
                kwargs["rescale"] = f"{percentile_2},{percentile_98}"

    else:
        data = requests.get(url).json()
        if "mosaicjson" in data:
            mosaic_json = True

        if isinstance(bands, str):
            bands = bands.split(",")
        if isinstance(assets, str):
            assets = assets.split(",")

        if assets is None:
            if bands is not None:
                assets = bands
            else:
                bnames = stac_bands(url)
                if isinstance(bnames, list):
                    if len(bnames) >= 3:
                        assets = bnames[0:3]
                    else:
                        assets = bnames[0]
                else:
                    assets = None

        else:
            if bands is not None:
                kwargs["asset_bidx"] = bands
        if assets is not None:
            kwargs["assets"] = assets

            if len(kwargs["assets"]) == 1 and ("colormap" not in kwargs):
                cog_url = get_cog_link_from_stac_item(url)
                colormap = _get_image_colormap(cog_url)
                if colormap is not None:
                    kwargs["colormap"] = colormap

        if (
            (assets is not None)
            and ("asset_expression" not in kwargs)
            and ("expression" not in kwargs)
            and ("rescale" not in kwargs)
            and ("colormap" not in kwargs)
        ):
            try:
                stats = stac_stats(
                    url=url,
                    assets=assets,
                    titiler_endpoint=titiler_endpoint,
                )
            except Exception as e:
                print(e)
                return None

            if "detail" not in stats:
                try:
                    percentile_2 = min([stats[s]["percentile_2"] for s in stats])
                    percentile_98 = max([stats[s]["percentile_98"] for s in stats])
                except:
                    percentile_2 = min(
                        [
                            stats[s][list(stats[s].keys())[0]]["percentile_2"]
                            for s in stats
                        ]
                    )
                    percentile_98 = max(
                        [
                            stats[s][list(stats[s].keys())[0]]["percentile_98"]
                            for s in stats
                        ]
                    )
                kwargs["rescale"] = f"{percentile_2},{percentile_98}"
            else:
                print(stats["detail"])  # When operation times out.

    TileMatrixSetId = "WebMercatorQuad"
    if "TileMatrixSetId" in kwargs.keys():
        TileMatrixSetId = kwargs["TileMatrixSetId"]
        kwargs.pop("TileMatrixSetId")

    if "colormap" in kwargs and isinstance(kwargs["colormap"], dict):
        kwargs["colormap"] = json.dumps(kwargs["colormap"])

    if mosaic_json:
        try:
            r = requests.get(
                f"{titiler_endpoint}/mosaicjson/{TileMatrixSetId}/tilejson.json",
                params=kwargs,
                timeout=10,
            ).json()
        except Exception as e:
            titiler_endpoint = "https://giswqs-titiler-endpoint.hf.space"
            r = requests.get(
                f"{titiler_endpoint}/mosaicjson/{TileMatrixSetId}/tilejson.json",
                params=kwargs,
                timeout=10,
            ).json()
    else:
        if isinstance(titiler_endpoint, str):
            try:
                r = requests.get(
                    f"{titiler_endpoint}/stac/{TileMatrixSetId}/tilejson.json",
                    params=kwargs,
                    timeout=10,
                ).json()
            except Exception as e:
                titiler_endpoint = "https://giswqs-titiler-endpoint.hf.space"
                r = requests.get(
                    f"{titiler_endpoint}/stac/{TileMatrixSetId}/tilejson.json",
                    params=kwargs,
                    timeout=10,
                ).json()
        else:
            r = requests.get(
                titiler_endpoint.url_for_stac_item(), params=kwargs, timeout=10
            ).json()

    # Check if the response contains tiles
    if "tiles" not in r:
        # Try to provide helpful error message from the API response
        if "detail" in r:
            error_msg = f"TiTiler endpoint error: {r['detail']}"
        elif isinstance(r, list) and len(r) > 0:
            # Handle validation errors (list of error dicts)
            if isinstance(r[0], dict) and "msg" in r[0]:
                errors = []
                for e in r:
                    loc = "->".join(str(x) for x in e.get("loc", []))
                    msg = e.get("msg", "")
                    errors.append(f"{loc}: {msg}")
                error_msg = f"TiTiler endpoint validation failed: {'; '.join(errors)}"
            else:
                error_msg = f"TiTiler endpoint returned error list: {r}"
        else:
            error_msg = f"TiTiler endpoint returned unexpected response (missing 'tiles' key): {r}"
        raise ValueError(error_msg)

    tiles = r["tiles"][0]

    # Convert 127.0.0.1 to localhost for Google Colab browser access
    if "google.colab" in sys.modules and "127.0.0.1" in tiles:
        tiles = tiles.replace("http://127.0.0.1", "https://localhost")

    return tiles

zarr_bounds(url, variable=None, titiler_endpoint=None, group=None, decode_times=False, **kwargs)

Get the bounds of a Zarr dataset.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a Zarr dataset.	required
variable	str	The variable name. Defaults to None.	None
titiler_endpoint	str	TiTiler endpoint that supports xarray/Zarr. If not provided, the function first checks the TITILER_XARRAY_ENDPOINT environment variable; if that is not set, it falls back to the default endpoint returned by :func:check_titiler_endpoint.	None
group	str	The Zarr group path within the dataset. Defaults to None.	None
decode_times	bool	Whether to decode times in the dataset. Defaults to False.	False
**kwargs		Additional arguments to pass to the titiler endpoint.	{}

Returns:

Name	Type	Description
list	list	Bounds as [minx, miny, maxx, maxy] in WGS84.

Source code in leafmap/stac.py

def zarr_bounds(
    url: str,
    variable: Optional[str] = None,
    titiler_endpoint: Optional[str] = None,
    group: Optional[str] = None,
    decode_times: bool = False,
    **kwargs,
) -> list:
    """Get the bounds of a Zarr dataset.

    Args:
        url (str): HTTP URL to a Zarr dataset.
        variable (str, optional): The variable name. Defaults to None.
        titiler_endpoint (str, optional): TiTiler endpoint that supports xarray/Zarr.
            If not provided, the function first checks the ``TITILER_XARRAY_ENDPOINT``
            environment variable; if that is not set, it falls back to the default
            endpoint returned by :func:`check_titiler_endpoint`.
        group (str, optional): The Zarr group path within the dataset.
            Defaults to None.
        decode_times (bool, optional): Whether to decode times in the dataset.
            Defaults to False.
        **kwargs: Additional arguments to pass to the titiler endpoint.

    Returns:
        list: Bounds as [minx, miny, maxx, maxy] in WGS84.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    # Check for xarray-specific endpoint
    if titiler_endpoint is None:
        titiler_endpoint = os.environ.get("TITILER_XARRAY_ENDPOINT")

    if titiler_endpoint is None:
        titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    params = {"url": url, "decode_times": str(decode_times).lower()}
    if variable is not None:
        params["variable"] = variable
    if group is not None:
        params["group"] = group
    params.update(kwargs)

    try:
        # Use /info endpoint since /bounds may not exist in titiler-xarray
        r = requests.get(
            f"{titiler_endpoint}/md/info",
            params=params,
            timeout=30,
        ).json()
        return r.get("bounds", None)
    except Exception as e:
        print(f"Error getting Zarr bounds: {e}")
        return None

zarr_info(url, variable=None, titiler_endpoint=None, group=None, decode_times=False, **kwargs)

Get information about a Zarr dataset.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a Zarr dataset.	required
variable	str	The variable name. Required by titiler-xarray. Defaults to None.	None
titiler_endpoint	str	TiTiler endpoint that supports xarray/Zarr. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
group	str	The Zarr group path within the dataset. Defaults to None.	None
decode_times	bool	Whether to decode times in the dataset. Defaults to False.	False
**kwargs		Additional arguments to pass to the titiler endpoint.	{}

Returns:

Name	Type	Description
dict	dict	Information about the Zarr dataset including bounds, CRS, and variables.

Source code in leafmap/stac.py

def zarr_info(
    url: str,
    variable: Optional[str] = None,
    titiler_endpoint: Optional[str] = None,
    group: Optional[str] = None,
    decode_times: bool = False,
    **kwargs,
) -> dict:
    """Get information about a Zarr dataset.

    Args:
        url (str): HTTP URL to a Zarr dataset.
        variable (str, optional): The variable name. Required by titiler-xarray.
            Defaults to None.
        titiler_endpoint (str, optional): TiTiler endpoint that supports xarray/Zarr.
            Defaults to "https://giswqs-titiler-endpoint.hf.space".
        group (str, optional): The Zarr group path within the dataset.
            Defaults to None.
        decode_times (bool, optional): Whether to decode times in the dataset.
            Defaults to False.
        **kwargs: Additional arguments to pass to the titiler endpoint.

    Returns:
        dict: Information about the Zarr dataset including bounds, CRS, and variables.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    # Check for xarray-specific endpoint
    if titiler_endpoint is None:
        titiler_endpoint = os.environ.get("TITILER_XARRAY_ENDPOINT")

    if titiler_endpoint is None:
        titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    params = {"url": url, "decode_times": str(decode_times).lower()}
    if variable is not None:
        params["variable"] = variable
    if group is not None:
        params["group"] = group
    params.update(kwargs)

    try:
        r = requests.get(
            f"{titiler_endpoint}/md/info",
            params=params,
            timeout=30,
        ).json()
        return r
    except Exception as e:
        print(f"Error getting Zarr info: {e}")
        return None

zarr_statistics(url, variable=None, titiler_endpoint=None, group=None, decode_times=False, **kwargs)

Get statistics for a Zarr dataset variable.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a Zarr dataset.	required
variable	str	The variable name. Required for multi-variable datasets.	None
titiler_endpoint	str	TiTiler endpoint that supports xarray/Zarr. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
group	str	The Zarr group path within the dataset. Defaults to None.	None
decode_times	bool	Whether to decode times in the dataset. Defaults to False.	False
**kwargs		Additional arguments to pass to the titiler endpoint.	{}

Returns:

Name	Type	Description
dict	dict	Statistics including min, max, mean, std, etc.

Source code in leafmap/stac.py

def zarr_statistics(
    url: str,
    variable: Optional[str] = None,
    titiler_endpoint: Optional[str] = None,
    group: Optional[str] = None,
    decode_times: bool = False,
    **kwargs,
) -> dict:
    """Get statistics for a Zarr dataset variable.

    Args:
        url (str): HTTP URL to a Zarr dataset.
        variable (str, optional): The variable name. Required for multi-variable datasets.
        titiler_endpoint (str, optional): TiTiler endpoint that supports xarray/Zarr.
            Defaults to "https://giswqs-titiler-endpoint.hf.space".
        group (str, optional): The Zarr group path within the dataset.
            Defaults to None.
        decode_times (bool, optional): Whether to decode times in the dataset.
            Defaults to False.
        **kwargs: Additional arguments to pass to the titiler endpoint.

    Returns:
        dict: Statistics including min, max, mean, std, etc.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    # Check for xarray-specific endpoint
    if titiler_endpoint is None:
        titiler_endpoint = os.environ.get("TITILER_XARRAY_ENDPOINT")

    if titiler_endpoint is None:
        titiler_endpoint = check_titiler_endpoint(titiler_endpoint)
    params = {"url": url, "decode_times": str(decode_times).lower()}
    if variable is not None:
        params["variable"] = variable
    if group is not None:
        params["group"] = group
    params.update(kwargs)

    try:
        r = requests.get(
            f"{titiler_endpoint}/md/statistics",
            params=params,
            timeout=30,
        ).json()
        return r
    except Exception as e:
        print(f"Error getting Zarr statistics: {e}")
        return None

zarr_tile(url, variable=None, titiler_endpoint=None, group=None, decode_times=False, time_index=0, **kwargs)

Get a tile layer URL from a Zarr dataset using titiler-xarray.

This function requires a TiTiler endpoint with titiler-xarray support. The default titiler endpoint does NOT support Zarr/xarray datasets. You need to either: 1. Set up your own titiler-xarray endpoint 2. Use leafmap.run_titiler_xarray() to start a local server

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a Zarr dataset, e.g., https://example.com/data.zarr or s3://bucket/data.zarr	required
variable	str	The variable name to visualize. Required for multi-variable datasets. Defaults to None.	None
titiler_endpoint	str	TiTiler endpoint that supports xarray/Zarr. Must have titiler-xarray installed. Defaults to environment variable TITILER_XARRAY_ENDPOINT or prompts user to start a local server.	None
group	str	The Zarr group path within the dataset. Defaults to None.	None
decode_times	bool	Whether to decode times in the dataset. Defaults to False.	False
time_index	int	Index of the time dimension to visualize. Required for datasets with a time dimension. Defaults to 0 (first time step). Set to None to disable time selection (for datasets without time dimension).	0
**kwargs	Any	Additional arguments to pass to the titiler endpoint. Common options include: - rescale (str): Comma-delimited Min,Max range (e.g., "0,255") - colormap_name (str): Name of a colormap (e.g., "viridis") - colormap (str): JSON encoded custom colormap - nodata (float): Nodata value - sel (str): Dimension selection in format "dim=value" (e.g., "time=0") - TileMatrixSetId (str): Tile matrix set ID (default: "WebMercatorQuad")	{}

Returns:

Name	Type	Description
str	str	The Zarr tile layer URL, or None if the endpoint doesn't support Zarr.

Example

Start local titiler-xarray server first

endpoint = leafmap.run_titiler_xarray() url = "https://example.com/data.zarr" tile_url = zarr_tile(url, variable="temperature", titiler_endpoint=endpoint)

Source code in leafmap/stac.py

def zarr_tile(
    url: str,
    variable: Optional[str] = None,
    titiler_endpoint: Optional[str] = None,
    group: Optional[str] = None,
    decode_times: bool = False,
    time_index: Optional[int] = 0,
    **kwargs,
) -> str:
    """Get a tile layer URL from a Zarr dataset using titiler-xarray.

    This function requires a TiTiler endpoint with titiler-xarray support.
    The default titiler endpoint does NOT support Zarr/xarray datasets.
    You need to either:
    1. Set up your own titiler-xarray endpoint
    2. Use `leafmap.run_titiler_xarray()` to start a local server

    Args:
        url (str): HTTP URL to a Zarr dataset, e.g., https://example.com/data.zarr
            or s3://bucket/data.zarr
        variable (str, optional): The variable name to visualize. Required for
            multi-variable datasets. Defaults to None.
        titiler_endpoint (str, optional): TiTiler endpoint that supports xarray/Zarr.
            Must have titiler-xarray installed. Defaults to environment variable
            TITILER_XARRAY_ENDPOINT or prompts user to start a local server.
        group (str, optional): The Zarr group path within the dataset.
            Defaults to None.
        decode_times (bool, optional): Whether to decode times in the dataset.
            Defaults to False.
        time_index (int, optional): Index of the time dimension to visualize.
            Required for datasets with a time dimension. Defaults to 0 (first time step).
            Set to None to disable time selection (for datasets without time dimension).
        **kwargs (Any): Additional arguments to pass to the titiler endpoint.
            Common options include:
            - rescale (str): Comma-delimited Min,Max range (e.g., "0,255")
            - colormap_name (str): Name of a colormap (e.g., "viridis")
            - colormap (str): JSON encoded custom colormap
            - nodata (float): Nodata value
            - sel (str): Dimension selection in format "dim=value" (e.g., "time=0")
            - TileMatrixSetId (str): Tile matrix set ID (default: "WebMercatorQuad")

    Returns:
        str: The Zarr tile layer URL, or None if the endpoint doesn't support Zarr.

    Example:
        >>> # Start local titiler-xarray server first
        >>> endpoint = leafmap.run_titiler_xarray()
        >>> url = "https://example.com/data.zarr"
        >>> tile_url = zarr_tile(url, variable="temperature", titiler_endpoint=endpoint)
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    # Check for xarray-specific endpoint
    if titiler_endpoint is None:
        titiler_endpoint = os.environ.get("TITILER_XARRAY_ENDPOINT")

    if titiler_endpoint is None:
        titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    kwargs["url"] = url

    if variable is not None:
        kwargs["variable"] = variable

    if group is not None:
        kwargs["group"] = group

    kwargs["decode_times"] = str(decode_times).lower()

    # Add time selection if time_index is specified
    # This is required for datasets with a time dimension
    if time_index is not None and "sel" not in kwargs:
        kwargs["sel"] = f"time={time_index}"

    if "palette" in kwargs:
        kwargs["colormap_name"] = kwargs["palette"].lower()
        del kwargs["palette"]

    # Ensure colormap_name is lowercase (required by titiler)
    if "colormap_name" in kwargs and kwargs["colormap_name"]:
        kwargs["colormap_name"] = kwargs["colormap_name"].lower()

    TileMatrixSetId = "WebMercatorQuad"
    if "TileMatrixSetId" in kwargs:
        TileMatrixSetId = kwargs["TileMatrixSetId"]
        kwargs.pop("TileMatrixSetId")

    try:
        r = requests.get(
            f"{titiler_endpoint}/md/{TileMatrixSetId}/tilejson.json",
            params=kwargs,
            timeout=30,
        ).json()

        # Check for various error conditions
        if "detail" in r:
            detail = r["detail"]
            if detail == "Not Found":
                print(
                    "Error: The titiler endpoint does not support Zarr/xarray datasets.\n"
                    "The '/md/' endpoint is not available. Please:\n"
                    "  1. Use a titiler endpoint with titiler-xarray installed, or\n"
                    "  2. Start a local titiler-xarray server with: leafmap.run_titiler_xarray()\n"
                    "  3. Set TITILER_XARRAY_ENDPOINT environment variable"
                )
                return None
            else:
                # Handle validation errors or other API errors
                print(f"Error from titiler endpoint: {detail}")
                return None

        if "tiles" not in r:
            print(f"Unexpected response from titiler: {r}")
            return None

        tiles = r["tiles"][0]
        if titiler_endpoint.startswith("https://") and tiles.startswith("http://"):
            tiles = tiles.replace("http://", "https://")

        # Convert 127.0.0.1 to localhost for Google Colab browser access
        if "google.colab" in sys.modules and "127.0.0.1" in tiles:
            tiles = tiles.replace("http://127.0.0.1", "https://localhost")

        return tiles
    except KeyError as e:
        print(f"Error getting Zarr tile URL: missing key {e}")
        return None
    except Exception as e:
        print(f"Error getting Zarr tile URL: {e}")
        return None

zarr_variables(url, titiler_endpoint=None, group=None, decode_times=False, **kwargs)

Get the list of variables from a Zarr dataset.

Parameters:

Name	Type	Description	Default
url	str	HTTP URL to a Zarr dataset.	required
titiler_endpoint	str	TiTiler endpoint that supports xarray/Zarr. Defaults to "https://giswqs-titiler-endpoint.hf.space".	None
group	str	The Zarr group path within the dataset. Defaults to None.	None
decode_times	bool	Whether to decode times in the dataset. Defaults to False.	False
**kwargs		Additional arguments to pass to the titiler endpoint.	{}

Returns:

Name	Type	Description
list	list	List of variable names in the Zarr dataset.

Source code in leafmap/stac.py

def zarr_variables(
    url: str,
    titiler_endpoint: Optional[str] = None,
    group: Optional[str] = None,
    decode_times: bool = False,
    **kwargs,
) -> list:
    """Get the list of variables from a Zarr dataset.

    Args:
        url (str): HTTP URL to a Zarr dataset.
        titiler_endpoint (str, optional): TiTiler endpoint that supports xarray/Zarr.
            Defaults to "https://giswqs-titiler-endpoint.hf.space".
        group (str, optional): The Zarr group path within the dataset.
            Defaults to None.
        decode_times (bool, optional): Whether to decode times in the dataset.
            Defaults to False.
        **kwargs: Additional arguments to pass to the titiler endpoint.

    Returns:
        list: List of variable names in the Zarr dataset.
    """
    if os.environ.get("USE_MKDOCS") is not None:
        return None

    # Check for xarray-specific endpoint
    if titiler_endpoint is None:
        titiler_endpoint = os.environ.get("TITILER_XARRAY_ENDPOINT")

    if titiler_endpoint is None:
        titiler_endpoint = check_titiler_endpoint(titiler_endpoint)

    params = {"url": url, "decode_times": str(decode_times).lower()}
    if group is not None:
        params["group"] = group
    params.update(kwargs)

    try:
        r = requests.get(
            f"{titiler_endpoint}/md/variables",
            params=params,
            timeout=30,
        ).json()
        return r
    except Exception as e:
        print(f"Error getting Zarr variables: {e}")
        return None