Skip to content

rslearn.utils.stac

stac

STAC API client.

StacAsset dataclass

A STAC asset.

Source code in rslearn/utils/stac.py 
15
16
17
18
19
20
21
22
@dataclass(frozen=True)
class StacAsset:
    """A STAC asset."""

    href: str
    title: str | None
    type: str | None
    roles: list[str] | None

StacItem dataclass

A STAC item.

Source code in rslearn/utils/stac.py 
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass(frozen=True)
class StacItem:
    """A STAC item."""

    id: str
    properties: dict[str, Any]
    collection: str | None
    bbox: Bbox | None
    geometry: dict[str, Any] | None
    assets: dict[str, StacAsset] | None
    time_range: tuple[datetime, datetime] | None

    @classmethod
    def from_dict(cls, item: dict[str, Any]) -> "StacItem":
        """Create a STAC item from the item dict returned from API."""
        properties = item.get("properties", {})

        # Parse bbox.
        bbox: Bbox | None = None
        if "bbox" in item:
            if len(item["bbox"]) != 4:
                raise NotImplementedError(
                    f"got bbox with {len(item['bbox'])} coordinates but only 4 coordinates is implemented"
                )
            bbox = tuple(item["bbox"])

        # Parse assets.
        assets: dict[str, StacAsset] = {}
        for name, asset in item.get("assets", {}).items():
            assets[name] = StacAsset(
                href=asset["href"],
                title=asset.get("title"),
                type=asset.get("type"),
                roles=asset.get("roles"),
            )

        # Parse time range.
        time_range: tuple[datetime, datetime] | None = None
        if "start_datetime" in properties and "end_datetime" in properties:
            time_range = (
                datetime.fromisoformat(properties["start_datetime"]),
                datetime.fromisoformat(properties["end_datetime"]),
            )
        elif "datetime" in properties:
            ts = datetime.fromisoformat(properties["datetime"])
            time_range = (ts, ts)

        return cls(
            id=item["id"],
            properties=properties,
            collection=item.get("collection"),
            bbox=bbox,
            geometry=item.get("geometry"),
            assets=assets,
            time_range=time_range,
        )

from_dict classmethod 

from_dict(item: dict[str, Any]) -> StacItem

Create a STAC item from the item dict returned from API.

Source code in rslearn/utils/stac.py 
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def from_dict(cls, item: dict[str, Any]) -> "StacItem":
    """Create a STAC item from the item dict returned from API."""
    properties = item.get("properties", {})

    # Parse bbox.
    bbox: Bbox | None = None
    if "bbox" in item:
        if len(item["bbox"]) != 4:
            raise NotImplementedError(
                f"got bbox with {len(item['bbox'])} coordinates but only 4 coordinates is implemented"
            )
        bbox = tuple(item["bbox"])

    # Parse assets.
    assets: dict[str, StacAsset] = {}
    for name, asset in item.get("assets", {}).items():
        assets[name] = StacAsset(
            href=asset["href"],
            title=asset.get("title"),
            type=asset.get("type"),
            roles=asset.get("roles"),
        )

    # Parse time range.
    time_range: tuple[datetime, datetime] | None = None
    if "start_datetime" in properties and "end_datetime" in properties:
        time_range = (
            datetime.fromisoformat(properties["start_datetime"]),
            datetime.fromisoformat(properties["end_datetime"]),
        )
    elif "datetime" in properties:
        ts = datetime.fromisoformat(properties["datetime"])
        time_range = (ts, ts)

    return cls(
        id=item["id"],
        properties=properties,
        collection=item.get("collection"),
        bbox=bbox,
        geometry=item.get("geometry"),
        assets=assets,
        time_range=time_range,
    )

StacClient

Limited functionality client for STAC APIs.

Source code in rslearn/utils/stac.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class StacClient:
    """Limited functionality client for STAC APIs."""

    def __init__(self, endpoint: str):
        """Create a new StacClient.

        Args:
            endpoint: the STAC endpoint (base URL)
        """
        self.endpoint = endpoint
        self.session = requests.Session()

    def search(
        self,
        collections: list[str] | None = None,
        bbox: Bbox | None = None,
        intersects: dict[str, Any] | None = None,
        date_time: datetime | tuple[datetime, datetime] | None = None,
        ids: list[str] | None = None,
        limit: int | None = None,
        query: dict[str, Any] | None = None,
        sortby: list[dict[str, str]] | None = None,
    ) -> list[StacItem]:
        """Execute a STAC item search.

        We use the JSON POST API. Pagination is handled so the returned items are
        concatenated across all available pages.

        Args:
            collections: only search within the provided collection(s).
            bbox: only return features intersecting the provided bounding box.
            intersects: only return features intersecting this GeoJSON geometry.
            date_time: only return features that have a temporal property intersecting
                the provided time range or timestamp.
            ids: only return the provided item IDs.
            limit: number of items per page. We will read all the pages.
            query: query dict, if STAC query extension is supported by this API. See
                https://github.com/stac-api-extensions/query.
            sortby: list of sort specifications, e.g. [{"field": "id", "direction": "asc"}].

        Returns:
            list of matching STAC items.
        """
        # Build JSON request data.
        request_data: dict[str, Any] = {}
        if collections is not None:
            request_data["collections"] = collections
        if bbox is not None:
            request_data["bbox"] = bbox
        if intersects is not None:
            request_data["intersects"] = intersects
        if date_time is not None:
            if isinstance(date_time, tuple):
                start_time = date_time[0].isoformat().replace("+00:00", "Z")
                end_time = date_time[1].isoformat().replace("+00:00", "Z")
                request_data["datetime"] = f"{start_time}/{end_time}"
            else:
                request_data["datetime"] = date_time.isoformat().replace("+00:00", "Z")
        if ids is not None:
            request_data["ids"] = ids
        if limit is not None:
            request_data["limit"] = limit
        if query is not None:
            request_data["query"] = query
        if sortby is not None:
            request_data["sortby"] = sortby

        # Handle pagination.
        cur_url = self.endpoint + "/search"
        cur_method = "POST"
        items: list[StacItem] = []
        while True:
            logger.debug(
                "Reading STAC items from %s with request data %s", cur_url, request_data
            )
            if cur_method == "POST":
                response = self.session.post(url=cur_url, json=request_data)
            elif cur_method == "GET":
                # GET pagination links are complete URLs from the previous response,
                # so request_data is intentionally not passed separately.
                response = self.session.get(url=cur_url)
            else:  # pragma: no cover - defensive check
                raise ValueError(f"unsupported STAC pagination method {cur_method!r}")
            response.raise_for_status()
            data = response.json()
            for item_dict in data["features"]:
                items.append(StacItem.from_dict(item_dict))

            next_link = None
            next_method = cur_method
            next_request_data: dict[str, Any] = {}
            for link in data.get("links", []):
                if "rel" not in link or link["rel"] != "next":
                    continue
                next_link = link["href"]
                next_method = link.get("method", cur_method).upper()
                if next_method == "POST":
                    next_request_data = link.get("body", {})
                elif next_method == "GET":
                    next_request_data = {}
                else:
                    raise ValueError(
                        f"unsupported STAC pagination method {next_method!r}"
                    )
                break

            if next_link is None:
                break

            cur_url = next_link
            cur_method = next_method
            request_data = next_request_data

        return items

search 

search(collections: list[str] | None = None, bbox: Bbox | None = None, intersects: dict[str, Any] | None = None, date_time: datetime | tuple[datetime, datetime] | None = None, ids: list[str] | None = None, limit: int | None = None, query: dict[str, Any] | None = None, sortby: list[dict[str, str]] | None = None) -> list[StacItem]

Execute a STAC item search.

We use the JSON POST API. Pagination is handled so the returned items are concatenated across all available pages.

Parameters:

Name Type Description Default
collections list[str] | None

only search within the provided collection(s).

 None
bbox Bbox | None

only return features intersecting the provided bounding box.

 None
intersects dict[str, Any] | None

only return features intersecting this GeoJSON geometry.

 None
date_time datetime | tuple[datetime, datetime] | None

only return features that have a temporal property intersecting the provided time range or timestamp.

 None
ids list[str] | None

only return the provided item IDs.

 None
limit int | None

number of items per page. We will read all the pages.

 None
query dict[str, Any] | None

query dict, if STAC query extension is supported by this API. See https://github.com/stac-api-extensions/query.

 None
sortby list[dict[str, str]] | None

list of sort specifications, e.g. [{"field": "id", "direction": "asc"}].

 None

Returns:

Type Description
list[StacItem]

list of matching STAC items.
Source code in rslearn/utils/stac.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def search(
    self,
    collections: list[str] | None = None,
    bbox: Bbox | None = None,
    intersects: dict[str, Any] | None = None,
    date_time: datetime | tuple[datetime, datetime] | None = None,
    ids: list[str] | None = None,
    limit: int | None = None,
    query: dict[str, Any] | None = None,
    sortby: list[dict[str, str]] | None = None,
) -> list[StacItem]:
    """Execute a STAC item search.

    We use the JSON POST API. Pagination is handled so the returned items are
    concatenated across all available pages.

    Args:
        collections: only search within the provided collection(s).
        bbox: only return features intersecting the provided bounding box.
        intersects: only return features intersecting this GeoJSON geometry.
        date_time: only return features that have a temporal property intersecting
            the provided time range or timestamp.
        ids: only return the provided item IDs.
        limit: number of items per page. We will read all the pages.
        query: query dict, if STAC query extension is supported by this API. See
            https://github.com/stac-api-extensions/query.
        sortby: list of sort specifications, e.g. [{"field": "id", "direction": "asc"}].

    Returns:
        list of matching STAC items.
    """
    # Build JSON request data.
    request_data: dict[str, Any] = {}
    if collections is not None:
        request_data["collections"] = collections
    if bbox is not None:
        request_data["bbox"] = bbox
    if intersects is not None:
        request_data["intersects"] = intersects
    if date_time is not None:
        if isinstance(date_time, tuple):
            start_time = date_time[0].isoformat().replace("+00:00", "Z")
            end_time = date_time[1].isoformat().replace("+00:00", "Z")
            request_data["datetime"] = f"{start_time}/{end_time}"
        else:
            request_data["datetime"] = date_time.isoformat().replace("+00:00", "Z")
    if ids is not None:
        request_data["ids"] = ids
    if limit is not None:
        request_data["limit"] = limit
    if query is not None:
        request_data["query"] = query
    if sortby is not None:
        request_data["sortby"] = sortby

    # Handle pagination.
    cur_url = self.endpoint + "/search"
    cur_method = "POST"
    items: list[StacItem] = []
    while True:
        logger.debug(
            "Reading STAC items from %s with request data %s", cur_url, request_data
        )
        if cur_method == "POST":
            response = self.session.post(url=cur_url, json=request_data)
        elif cur_method == "GET":
            # GET pagination links are complete URLs from the previous response,
            # so request_data is intentionally not passed separately.
            response = self.session.get(url=cur_url)
        else:  # pragma: no cover - defensive check
            raise ValueError(f"unsupported STAC pagination method {cur_method!r}")
        response.raise_for_status()
        data = response.json()
        for item_dict in data["features"]:
            items.append(StacItem.from_dict(item_dict))

        next_link = None
        next_method = cur_method
        next_request_data: dict[str, Any] = {}
        for link in data.get("links", []):
            if "rel" not in link or link["rel"] != "next":
                continue
            next_link = link["href"]
            next_method = link.get("method", cur_method).upper()
            if next_method == "POST":
                next_request_data = link.get("body", {})
            elif next_method == "GET":
                next_request_data = {}
            else:
                raise ValueError(
                    f"unsupported STAC pagination method {next_method!r}"
                )
            break

        if next_link is None:
            break

        cur_url = next_link
        cur_method = next_method
        request_data = next_request_data

    return items