Skip to content

Description

Step purpose

Purpose: produce a STAC Catalog with a STAC Item describing the detected water body result.

This step is highlighted below:

graph TB style F stroke:#f66,stroke-width:3px subgraph Process STAC item A[STAC Item] -.-> B A[STAC Item] -.-> C A[STAC Item] == STAC Item URL ==> F subgraph scatter on bands B["crop(green)"]; C["crop(nir)"]; end B["crop(green)"] -.-> D[Normalized difference]; C["crop(nir)"] -.-> D[Normalized difference]; D -.-> E[Otsu threshold] end E == otsu.tif ==> F[Create STAC Catalog] F == "catalog.json/item.json/asset otsu.tif" ==> G[(storage)]

Code

The stac script is a command-line tool for creating a SpatioTemporal Asset Catalog (STAC) catalog containing detected water bodies.

It takes the STAC Item URLs and corresponding water body GeoTIFFs as input, creates STAC Items for each water body, and organizes them into a STAC Catalog.

The script utilizes the click, pystac, rio_stac, and loguru libraries.

Here's an overview of what the script does:

  • It defines a command-line interface using the click library. The script expects multiple input STAC Item URLs and corresponding water body GeoTIFFs as arguments.

  • The to_stac function is the main entry point. It creates a new STAC catalog and iterates through the provided STAC Item URLs and water body GeoTIFFs.

  • For each input pair (STAC Item URL and GeoTIFF), it reads the STAC Item, creates a directory with the same name as the item's ID, and copies the water body GeoTIFF into that directory.

  • It uses the rio_stac library to create a STAC Item for the water body. This includes specifying the source GeoTIFF, input date/time, asset roles, asset href, and more.

  • The created STAC Items are added to the STAC catalog.

  • After processing all input pairs, it saves the STAC catalog to the root directory, specifying it as a self-contained catalog with the catalog type set to pystac.CatalogType.SELF_CONTAINED.

The script will create a STAC catalog containing the detected water bodies and save it in the current directory.

The script is executable as a command-line tool as its usage is:

Usage: app.py [OPTIONS]

  Creates a STAC catalog with the water bodies

Options:
  --input-item TEXT  STAC Item URL  [required]
  --water-body TEXT  Water body geotiff  [required]
  --help             Show this message and exit.

The Python code is provided here:

water-bodies/command-line-tools/stac/app.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
"""Creates a STAC catalog with the detected water bodies""" ""
import os
import shutil
import click
import pystac
import rio_stac
from loguru import logger


@click.command(
    short_help="Creates a STAC catalog",
    help="Creates a STAC catalog with the water bodies",
)
@click.option(
    "--input-item",
    "item_urls",
    help="STAC Item URL",
    required=True,
    multiple=True,
)
@click.option(
    "--water-body",
    "water_bodies",
    help="Water body geotiff",
    required=True,
    multiple=True,
)
def to_stac(item_urls, water_bodies):
    """Creates a STAC catalog with the detected water bodies"""

    logger.info(f"Creating a STAC Catalog for {' '.join(water_bodies)}")
    cat = pystac.Catalog(id="catalog", description="water-bodies")

    for index, item_url in enumerate(item_urls):
        if os.path.isdir(item_url):
            catalog = pystac.read_file(os.path.join(item_url, "catalog.json"))
            item = next(catalog.get_items())
        else:
            item = pystac.read_file(item_url)

        water_body = water_bodies[index]

        os.makedirs(item.id, exist_ok=True)
        shutil.copy(water_body, item.id)

        out_item = rio_stac.stac.create_stac_item(
            source=water_body,
            input_datetime=item.datetime,
            id=item.id,
            asset_roles=["data", "visual"],
            asset_href=os.path.basename(water_body),
            asset_name="data",
            with_proj=True,
            with_raster=False,
        )

        cat.add_items([out_item])

    cat.normalize_and_save(
        root_href="./", catalog_type=pystac.CatalogType.SELF_CONTAINED
    )
    logger.info("Done!")


if __name__ == "__main__":
    to_stac()