Skip to content

terrascope module

Terrascope STAC API integration for leafmap.

This module provides authentication and helper functions for accessing Terrascope data services, including: - OAuth2 token management with automatic refresh - STAC catalog search helpers - GDAL authentication for streaming COGs

Example

import leafmap.terrascope as terrascope terrascope.login() # Uses TERRASCOPE_USERNAME/PASSWORD env vars items = terrascope.search_ndvi(bbox=[5.0, 51.2, 5.1, 51.3], start="2025-05-01", end="2025-06-01") m = leafmap.Map() m.add_raster(items[0].assets["NDVI"].href, colormap="RdYlGn")

Note

Authentication uses OAuth2 Resource Owner Password Credentials Grant. Credentials are transmitted securely to Terrascope's SSO endpoint and are not stored locally. Only the resulting tokens are cached.

cleanup_tile_servers()

Kill stale localtileserver processes.

This is useful when switching between visualizations to avoid authentication errors from old tile server processes that have cached expired tokens.

Note

This uses pkill which is Unix-specific and will terminate all localtileserver processes for the current user, not just those started by this session.

Source code in leafmap/terrascope.py
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
def cleanup_tile_servers() -> None:
    """
    Kill stale localtileserver processes.

    This is useful when switching between visualizations to avoid
    authentication errors from old tile server processes that have
    cached expired tokens.

    Note:
        This uses pkill which is Unix-specific and will terminate all
        localtileserver processes for the current user, not just those
        started by this session.
    """
    try:
        subprocess.run(["pkill", "-f", "localtileserver"], capture_output=True)
    except Exception:
        # Ignore errors: pkill may not exist on all platforms (e.g., Windows),
        # or no matching processes may be found. This cleanup is best-effort.
        pass

create_time_layers(items, asset_key='NDVI', colormap='RdYlGn', vmin=0, vmax=250)

Create tile layers for time slider visualization.

Parameters:

Name Type Description Default
items list

List of pystac Item objects.

required
asset_key str

Asset key to use (default "NDVI").

'NDVI'
colormap str

Colormap name (default "RdYlGn").

'RdYlGn'
vmin float

Minimum value for colormap.

0
vmax float

Maximum value for colormap.

250

Returns:

Type Description
dict

Dictionary of {date_string: tile_layer} for use with Map.add_time_slider().

Example

items = terrascope.search_ndvi(bbox, start, end) layers = terrascope.create_time_layers(items) m = leafmap.Map() m.add_time_slider(layers)

Source code in leafmap/terrascope.py
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
def create_time_layers(
    items: list,
    asset_key: str = "NDVI",
    colormap: str = "RdYlGn",
    vmin: float = 0,
    vmax: float = 250,
) -> dict:
    """
    Create tile layers for time slider visualization.

    Args:
        items: List of pystac Item objects.
        asset_key: Asset key to use (default "NDVI").
        colormap: Colormap name (default "RdYlGn").
        vmin: Minimum value for colormap.
        vmax: Maximum value for colormap.

    Returns:
        Dictionary of {date_string: tile_layer} for use with Map.add_time_slider().

    Example:
        >>> items = terrascope.search_ndvi(bbox, start, end)
        >>> layers = terrascope.create_time_layers(items)
        >>> m = leafmap.Map()
        >>> m.add_time_slider(layers)
    """
    try:
        import leafmap
    except ImportError:
        raise ImportError("leafmap is required: pip install leafmap")

    layers = {}
    for item in items:
        if asset_key not in item.assets:
            continue
        date_str = item.datetime.strftime("%Y-%m-%d")
        tile_layer = leafmap.get_local_tile_layer(
            item.assets[asset_key].href,
            layer_name=date_str,
            colormap=colormap,
            vmin=vmin,
            vmax=vmax,
        )
        layers[date_str] = tile_layer

    return layers

get_asset_urls(items, asset_key='NDVI')

Extract asset URLs from a list of STAC items.

Parameters:

Name Type Description Default
items list

List of pystac Item objects.

required
asset_key str

Asset key to extract (default "NDVI").

'NDVI'

Returns:

Type Description
list[str]

List of asset URLs.

Source code in leafmap/terrascope.py
492
493
494
495
496
497
498
499
500
501
502
503
def get_asset_urls(items: list, asset_key: str = "NDVI") -> list[str]:
    """
    Extract asset URLs from a list of STAC items.

    Args:
        items: List of pystac Item objects.
        asset_key: Asset key to extract (default "NDVI").

    Returns:
        List of asset URLs.
    """
    return [item.assets[asset_key].href for item in items if asset_key in item.assets]

get_item_dates(items)

Extract dates from a list of STAC items.

Parameters:

Name Type Description Default
items list

List of pystac Item objects.

required

Returns:

Type Description
list[str]

List of date strings in "YYYY-MM-DD" format.

Source code in leafmap/terrascope.py
506
507
508
509
510
511
512
513
514
515
516
def get_item_dates(items: list) -> list[str]:
    """
    Extract dates from a list of STAC items.

    Args:
        items: List of pystac Item objects.

    Returns:
        List of date strings in "YYYY-MM-DD" format.
    """
    return [item.datetime.strftime("%Y-%m-%d") for item in items]

get_stac_client()

Get a pystac_client Client for the Terrascope STAC catalog.

Returns:

Type Description
Any

pystac_client.Client instance.

Source code in leafmap/terrascope.py
336
337
338
339
340
341
342
343
344
345
346
347
def get_stac_client() -> Any:
    """
    Get a pystac_client Client for the Terrascope STAC catalog.

    Returns:
        pystac_client.Client instance.
    """
    global _stac_client
    _check_stac_dependencies()
    if _stac_client is None:
        _stac_client = Client.open(STAC_URL)
    return _stac_client

get_token(username=None, password=None)

Get a valid Terrascope access token.

Attempts to get a token in this order: 1. Return cached token if still valid 2. Refresh using refresh token if available 3. Login with username/password

Parameters:

Name Type Description Default
username str | None

Terrascope username. Defaults to TERRASCOPE_USERNAME env var.

None
password str | None

Terrascope password. Defaults to TERRASCOPE_PASSWORD env var.

None

Returns:

Type Description
str

Valid access token string.

Raises:

Type Description
ValueError

If credentials are not provided and not in environment.

HTTPError

If authentication fails.

Source code in leafmap/terrascope.py
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def get_token(
    username: str | None = None,
    password: str | None = None,
) -> str:
    """
    Get a valid Terrascope access token.

    Attempts to get a token in this order:
    1. Return cached token if still valid
    2. Refresh using refresh token if available
    3. Login with username/password

    Args:
        username: Terrascope username. Defaults to TERRASCOPE_USERNAME env var.
        password: Terrascope password. Defaults to TERRASCOPE_PASSWORD env var.

    Returns:
        Valid access token string.

    Raises:
        ValueError: If credentials are not provided and not in environment.
        requests.HTTPError: If authentication fails.
    """
    global _token_cache
    _check_dependencies()

    with _token_lock:
        if not _token_cache:
            _load_cached_tokens()

        now = time.time()

        # Check if access token is still valid
        if _token_cache.get("access_expires_at", 0) > now:
            return _token_cache["access_token"]

        # Try to refresh if refresh token is still valid
        if _token_cache.get("refresh_expires_at", 0) > now:
            try:
                return _refresh_access_token(_token_cache["refresh_token"])
            except Exception:
                # Fall through to password auth
                pass

        # Need fresh login with credentials
        username = username or os.environ.get("TERRASCOPE_USERNAME")
        password = password or os.environ.get("TERRASCOPE_PASSWORD")
        if not username or not password:
            raise ValueError(
                "Terrascope credentials required. Either pass username/password "
                "or set TERRASCOPE_USERNAME and TERRASCOPE_PASSWORD environment "
                "variables."
            )
        return _get_token_with_password(username, password)

list_collections()

List available collections in the Terrascope STAC catalog.

Returns:

Type Description
list[str]

List of collection IDs.

Source code in leafmap/terrascope.py
350
351
352
353
354
355
356
357
358
def list_collections() -> list[str]:
    """
    List available collections in the Terrascope STAC catalog.

    Returns:
        List of collection IDs.
    """
    client = get_stac_client()
    return [c.id for c in client.get_collections()]

login(username=None, password=None, auto_refresh=True, quiet=False)

Authenticate with Terrascope and configure GDAL for COG streaming.

This sets up the necessary environment variables and header files for GDAL/rasterio to authenticate with Terrascope when streaming COGs.

Parameters:

Name Type Description Default
username str | None

Terrascope username. Defaults to TERRASCOPE_USERNAME env var.

None
password str | None

Terrascope password. Defaults to TERRASCOPE_PASSWORD env var.

None
auto_refresh bool

Start background thread to refresh token every 4 minutes.

True
quiet bool

Suppress status messages.

False

Returns:

Type Description
str

Access token string.

Example

import leafmap.terrascope as terrascope terrascope.login() Authenticated as: your_username Background token refresh started (every 4 min)

Source code in leafmap/terrascope.py
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
def login(
    username: str | None = None,
    password: str | None = None,
    auto_refresh: bool = True,
    quiet: bool = False,
) -> str:
    """
    Authenticate with Terrascope and configure GDAL for COG streaming.

    This sets up the necessary environment variables and header files for
    GDAL/rasterio to authenticate with Terrascope when streaming COGs.

    Args:
        username: Terrascope username. Defaults to TERRASCOPE_USERNAME env var.
        password: Terrascope password. Defaults to TERRASCOPE_PASSWORD env var.
        auto_refresh: Start background thread to refresh token every 4 minutes.
        quiet: Suppress status messages.

    Returns:
        Access token string.

    Example:
        >>> import leafmap.terrascope as terrascope
        >>> terrascope.login()
        Authenticated as: your_username
        Background token refresh started (every 4 min)
    """
    global _refresh_thread, _refresh_stop

    token = get_token(username, password)

    with _token_lock:
        _update_header_file(token)

    os.environ["GDAL_HTTP_HEADER_FILE"] = HEADER_FILE_PATH
    os.environ["GDAL_DISABLE_READDIR_ON_OPEN"] = "EMPTY_DIR"

    # Start background refresher if not already running
    if auto_refresh and (_refresh_thread is None or not _refresh_thread.is_alive()):
        _refresh_stop.clear()
        _refresh_thread = threading.Thread(target=_background_refresher, daemon=True)
        _refresh_thread.start()
        if not quiet:
            print("Background token refresh started (every 4 min)")

    if not quiet:
        display_username = username or os.environ.get("TERRASCOPE_USERNAME", "(cached)")
        print(f"Authenticated as: {display_username}")

    return token

logout()

Stop background token refresh and clear cached tokens.

Also unsets GDAL environment variables that were configured during login.

Source code in leafmap/terrascope.py
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def logout() -> None:
    """
    Stop background token refresh and clear cached tokens.

    Also unsets GDAL environment variables that were configured during login.
    """
    global _refresh_thread, _token_cache
    _refresh_stop.set()
    if _refresh_thread:
        _refresh_thread.join(timeout=1)
        _refresh_thread = None

    with _token_lock:
        _token_cache = {}
        if os.path.exists(TOKEN_CACHE_PATH):
            os.remove(TOKEN_CACHE_PATH)
        if os.path.exists(HEADER_FILE_PATH):
            os.remove(HEADER_FILE_PATH)

    # Unset GDAL environment variables configured in login()
    os.environ.pop("GDAL_HTTP_HEADER_FILE", None)
    os.environ.pop("GDAL_DISABLE_READDIR_ON_OPEN", None)

    print("Logged out from Terrascope")

search(collection, bbox=None, start=None, end=None, max_cloud_cover=None, limit=None, unique_dates=True, **kwargs)

Search for items in a Terrascope STAC collection.

Parameters:

Name Type Description Default
collection str

Collection ID (e.g., "terrascope-s2-ndvi-v2").

required
bbox list[float] | None

Bounding box [west, south, east, north] in WGS84.

None
start str | datetime | None

Start date (string "YYYY-MM-DD" or datetime).

None
end str | datetime | None

End date (string "YYYY-MM-DD" or datetime).

None
max_cloud_cover float | None

Maximum cloud cover percentage (0-100).

None
limit int | None

Maximum number of items to return.

None
unique_dates bool

If True, return only one item per unique date.

True
**kwargs

Additional arguments passed to pystac_client search.

{}

Returns:

Type Description
list

List of pystac Item objects, sorted by date.

Example

items = terrascope.search( ... collection="terrascope-s2-ndvi-v2", ... bbox=[5.0, 51.2, 5.1, 51.3], ... start="2025-05-01", ... end="2025-06-01", ... max_cloud_cover=10, ... )

Source code in leafmap/terrascope.py
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def search(
    collection: str,
    bbox: list[float] | None = None,
    start: str | datetime | None = None,
    end: str | datetime | None = None,
    max_cloud_cover: float | None = None,
    limit: int | None = None,
    unique_dates: bool = True,
    **kwargs,
) -> list:
    """
    Search for items in a Terrascope STAC collection.

    Args:
        collection: Collection ID (e.g., "terrascope-s2-ndvi-v2").
        bbox: Bounding box [west, south, east, north] in WGS84.
        start: Start date (string "YYYY-MM-DD" or datetime).
        end: End date (string "YYYY-MM-DD" or datetime).
        max_cloud_cover: Maximum cloud cover percentage (0-100).
        limit: Maximum number of items to return.
        unique_dates: If True, return only one item per unique date.
        **kwargs: Additional arguments passed to pystac_client search.

    Returns:
        List of pystac Item objects, sorted by date.

    Example:
        >>> items = terrascope.search(
        ...     collection="terrascope-s2-ndvi-v2",
        ...     bbox=[5.0, 51.2, 5.1, 51.3],
        ...     start="2025-05-01",
        ...     end="2025-06-01",
        ...     max_cloud_cover=10,
        ... )
    """
    _check_stac_dependencies()
    client = get_stac_client()

    # Parse dates
    datetime_filter = None
    if start or end:
        if isinstance(start, str):
            start = datetime.fromisoformat(start).replace(tzinfo=timezone.utc)
        if isinstance(end, str):
            end = datetime.fromisoformat(end).replace(tzinfo=timezone.utc)
        datetime_filter = [start, end]

    # Build filter for cloud cover
    filter_expr = None
    if max_cloud_cover is not None:
        filter_expr = {
            "op": "<=",
            "args": [{"property": "properties.eo:cloud_cover"}, max_cloud_cover],
        }

    # Execute search
    search_kwargs = {
        "collections": [collection],
        **kwargs,
    }
    if bbox:
        search_kwargs["bbox"] = bbox
    if datetime_filter:
        search_kwargs["datetime"] = datetime_filter
    if filter_expr:
        search_kwargs["filter"] = filter_expr
    if limit and not unique_dates:
        search_kwargs["limit"] = limit

    search_result = client.search(**search_kwargs)
    items = list(search_result.items())

    # Sort by date
    items.sort(key=lambda i: i.datetime)

    # Filter to unique dates if requested
    if unique_dates:
        unique_items = {}
        for item in items:
            date_str = item.datetime.strftime("%Y-%m-%d")
            if date_str not in unique_items:
                unique_items[date_str] = item
        items = list(unique_items.values())

    # Apply limit after unique filtering
    if limit and len(items) > limit:
        items = items[:limit]

    return items

search_ndvi(bbox, start, end, max_cloud_cover=10.0, **kwargs)

Search for Sentinel-2 NDVI products.

Convenience wrapper around search() for the terrascope-s2-ndvi-v2 collection.

Parameters:

Name Type Description Default
bbox list[float]

Bounding box [west, south, east, north] in WGS84.

required
start str | datetime

Start date (string "YYYY-MM-DD" or datetime).

required
end str | datetime

End date (string "YYYY-MM-DD" or datetime).

required
max_cloud_cover float

Maximum cloud cover percentage. Default 10%.

10.0
**kwargs

Additional arguments passed to search().

{}

Returns:

Type Description
list

List of pystac Item objects with NDVI assets.

Example

items = terrascope.search_ndvi( ... bbox=[5.0, 51.2, 5.1, 51.3], ... start="2025-05-01", ... end="2025-06-01", ... ) print(f"Found {len(items)} NDVI scenes")

Source code in leafmap/terrascope.py
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
def search_ndvi(
    bbox: list[float],
    start: str | datetime,
    end: str | datetime,
    max_cloud_cover: float = 10.0,
    **kwargs,
) -> list:
    """
    Search for Sentinel-2 NDVI products.

    Convenience wrapper around search() for the terrascope-s2-ndvi-v2 collection.

    Args:
        bbox: Bounding box [west, south, east, north] in WGS84.
        start: Start date (string "YYYY-MM-DD" or datetime).
        end: End date (string "YYYY-MM-DD" or datetime).
        max_cloud_cover: Maximum cloud cover percentage. Default 10%.
        **kwargs: Additional arguments passed to search().

    Returns:
        List of pystac Item objects with NDVI assets.

    Example:
        >>> items = terrascope.search_ndvi(
        ...     bbox=[5.0, 51.2, 5.1, 51.3],
        ...     start="2025-05-01",
        ...     end="2025-06-01",
        ... )
        >>> print(f"Found {len(items)} NDVI scenes")
    """
    return search(
        collection="terrascope-s2-ndvi-v2",
        bbox=bbox,
        start=start,
        end=end,
        max_cloud_cover=max_cloud_cover,
        **kwargs,
    )