-
Notifications
You must be signed in to change notification settings - Fork 0
/
stacrs.pyi
297 lines (258 loc) · 11.6 KB
/
stacrs.pyi
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
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
235
236
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
287
288
289
290
291
292
293
294
295
296
297
from typing import Any, Optional, Tuple
def migrate_href(href: str, version: Optional[str] = None) -> dict[str, Any]:
"""
Migrates a STAC dictionary at the given href to another version.
Migration can be as simple as updating the `stac_version` attribute, but
sometimes can be more complicated. For example, when migrating to v1.1.0,
[eo:bands and raster:bands should be consolidated to the new bands
structure](https://github.com/radiantearth/stac-spec/releases/tag/v1.1.0-beta.1).
See [the stac-rs
documentation](https://docs.rs/stac/latest/stac/enum.Version.html) for
supported versions.
Args:
href (str): The href to read the STAC object from
version (str | None): The version to migrate to. If not provided, the
value will be migrated to the latest stable version.
Examples:
>>> item = stacrs.migrate_href("examples/simple-item.json", "1.1.0-beta.1")
>>> assert item["stac_version"] == "1.1.0-beta.1"
"""
def migrate(value: dict[str, Any], version: Optional[str] = None) -> dict[str, Any]:
"""
Migrates a STAC dictionary to another version.
Migration can be as simple as updating the `stac_version` attribute, but
sometimes can be more complicated. For example, when migrating to v1.1.0,
[eo:bands and raster:bands should be consolidated to the new bands
structure](https://github.com/radiantearth/stac-spec/releases/tag/v1.1.0-beta.1).
See [the stac-rs
documentation](https://docs.rs/stac/latest/stac/enum.Version.html) for
supported versions.
Args:
value (dict[str, Any]): The STAC value to migrate
version (str | None): The version to migrate to. If not provided, the
value will be migrated to the latest stable version.
Returns:
dict[str, Any]: The migrated dictionary
Examples:
>>> with open("examples/simple-item.json") as f:
>>> item = json.load(f)
>>> item = stacrs.migrate(item, "1.1.0-beta.1")
>>> assert item["stac_version"] == "1.1.0-beta.1"
"""
def read(
href: str,
*,
format: str | None = None,
options: list[tuple[str, str]] | None = None,
) -> dict[str, Any]:
"""
Reads STAC from a href.
Args:
href (str): The href to write to
format (str | None): The output format to write. If not provided, will be
inferred from the href's extension.
options (list[tuple[str, str]] | None): Options for configuring an
object store, e.g. your AWS credentials.
Returns:
dict[str, Any]: The STAC value
Examples:
>>> item = stacrs.read("item.json")
"""
def search(
href: str,
*,
intersects: Optional[str | dict[str, Any]] = None,
ids: Optional[str | list[str]] = None,
collections: Optional[str | list[str]] = None,
max_items: Optional[int] = None,
limit: Optional[int] = None,
bbox: Optional[list[float]] = None,
datetime: Optional[str] = None,
include: Optional[str | list[str]] = None,
exclude: Optional[str | list[str]] = None,
sortby: Optional[str | list[str]] = None,
filter: Optional[str | dict[str, Any]] = None,
query: Optional[dict[str, Any]] = None,
use_duckdb: Optional[bool] = None,
) -> list[dict[str, Any]]:
"""
Searches a STAC API server.
Args:
href (str): The STAC API to search.
intersects (str | dict[str, Any] | GeoInterface | None): Searches items
by performing intersection between their geometry and provided GeoJSON
geometry.
ids (list[str] | None): Array of Item ids to return.
collections (list[str] | None): Array of one or more Collection IDs that
each matching Item must be in.
max_items (int | None): The maximum number of items to iterate through.
limit (int | None): The page size returned from the server. Use
`max_items` to actually limit the number of items returned from this
function.
bbox (list[float] | None): Requested bounding box.
datetime (str | None): Single date+time, or a range (`/` separator),
formatted to RFC 3339, section 5.6. Use double dots .. for open
date ranges.
include (list[str]] | None): fields to include in the response (see [the
extension
docs](https://github.com/stac-api-extensions/fields?tab=readme-ov-file#includeexclude-semantics))
for more on the semantics).
exclude (list[str]] | None): fields to exclude from the response (see [the
extension
docs](https://github.com/stac-api-extensions/fields?tab=readme-ov-file#includeexclude-semantics))
for more on the semantics).
sortby (list[str] | None): Fields by which to sort results (use `-field` to sort descending).
filter (str | dict[str, Any] | none): CQL2 filter expression. Strings
will be interpreted as cql2-text, dictionaries as cql2-json.
query (dict[str, Any] | None): Additional filtering based on properties.
It is recommended to use filter instead, if possible.
use_duckdb (bool | None): Query with DuckDB. If None and the href has a
'parquet' or 'geoparquet' extension, will be set to True. Defaults
to None.
Returns:
list[dict[str, Any]]: A list of the returned STAC items.
Examples:
>>> items = stacrs.search(
... "https://landsatlook.usgs.gov/stac-server",
... collections=["landsat-c2l2-sr"],
... intersects={"type": "Point", "coordinates": [-105.119, 40.173]},
... sortby="-properties.datetime",
... max_items=1,
... )
"""
def search_to(
outfile: str,
href: str,
*,
intersects: Optional[str | dict[str, Any]] = None,
ids: Optional[str | list[str]] = None,
collections: Optional[str | list[str]] = None,
max_items: Optional[int] = None,
limit: Optional[int] = None,
bbox: Optional[list[float]] = None,
datetime: Optional[str] = None,
include: Optional[str | list[str]] = None,
exclude: Optional[str | list[str]] = None,
sortby: Optional[str | list[str]] = None,
filter: Optional[str | dict[str, Any]] = None,
query: Optional[dict[str, Any]] = None,
format: Optional[str] = None,
options: Optional[list[Tuple[str, str]]] = None,
use_duckdb: Optional[bool] = None,
) -> int:
"""
Searches a STAC API server and saves the result to an output file.
Args:
outfile (str): The output href. This can be a local file path, or any
url scheme supported by [stac::object_store::write].
href (str): The STAC API to search.
intersects (str | dict[str, Any] | GeoInterface | None): Searches items
by performing intersection between their geometry and provided GeoJSON
geometry.
ids (list[str] | None): Array of Item ids to return.
collections (list[str] | None): Array of one or more Collection IDs that
each matching Item must be in.
max_items (int | None): The maximum number of items to iterate through.
limit (int | None): The page size returned from the server. Use
`max_items` to actually limit the number of items returned from this
function.
bbox (list[float] | None): Requested bounding box.
datetime (str | None): Single date+time, or a range ('/' separator),
formatted to RFC 3339, section 5.6. Use double dots .. for open
date ranges.
include (list[str]] | None): fields to include in the response (see [the
extension
docs](https://github.com/stac-api-extensions/fields?tab=readme-ov-file#includeexclude-semantics))
for more on the semantics).
exclude (list[str]] | None): fields to exclude from the response (see [the
extension
docs](https://github.com/stac-api-extensions/fields?tab=readme-ov-file#includeexclude-semantics))
for more on the semantics).
sortby (list[str] | None): Fields by which to sort results (use `-field` to sort descending).
filter (str | dict[str, Any] | none): CQL2 filter expression. Strings
will be interpreted as cql2-text, dictionaries as cql2-json.
query (dict[str, Any] | None): Additional filtering based on properties.
It is recommended to use filter instead, if possible.
format (str | None): The output format. If none, will be inferred from
the outfile extension, and if that fails will fall back to compact JSON.
options (list[tuple[str, str]] | None): Configuration values to pass to the object store backend.
use_duckdb (bool | None): Query with DuckDB. If None and the href has a
'parquet' or 'geoparquet' extension, will be set to True. Defaults
to None.
Returns:
list[dict[str, Any]]: A list of the returned STAC items.
Examples:
>>> items = stacrs.search_to("out.parquet",
... "https://landsatlook.usgs.gov/stac-server",
... collections=["landsat-c2l2-sr"],
... intersects={"type": "Point", "coordinates": [-105.119, 40.173]},
... sortby="-properties.datetime",
... max_items=1,
... )
"""
def validate_href(href: str) -> None:
"""
Validates a single href with json-schema.
Args:
href (str): The href of the STAC value to validate
Raises:
Exception: On any input/output error, or on a validation error
Examples:
>>> stacrs.validate_href("examples/simple-item.json")
>>> stacrs.validate_href("data/invalid-item.json")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
Exception: Validation errors: "collection" is a required property
"""
def validate(value: dict[str, Any]) -> None:
"""
Validates a STAC dictionary with json-schema.
Args:
value (dict[str, Any]): The STAC value to validate
Raises:
Exception: On a validation error
Examples:
>>> with open("examples/simple-item.json") as f:
>>> data = json.load(f)
>>> stacrs.validate(data)
"""
def write(
href: str,
value: dict[str, Any] | list[dict[str, Any]],
*,
format: str | None = None,
options: list[tuple[str, str]] | None = None,
) -> dict[str, str] | None:
"""
Writes STAC to a href.
Args:
href (str): The href to write to
value (dict[str, Any] | list[dict[str, Any]]): The value to write. This
can be a STAC dictionary or a list of items.
format (str | None): The output format to write. If not provided, will be
inferred from the href's extension.
options (list[tuple[str, str]] | None): Options for configuring an
object store, e.g. your AWS credentials.
Returns:
dict[str, str] | None: The result of putting data into an object store,
e.g. the e_tag and the version. None is returned if the file was written
locally.
Examples:
>>> with open("items.json") as f:
... items = json.load(f)
>>> stacrs.write("items.parquet", items)
"""
def version(name: str | None = None) -> str | None:
"""
Returns this package's version, or the version of a upstream.
Args:
name (str | None): The name of the upstream version to return. Valid
values are "stac", "stac-api", "stac-duckdb", or "duckdb".
Returns:
str: The version, or None if the name is not recognized as an upstream.
Examples:
>>> stacrs.version()
"0.2.0"
>>> stacrs.version("duckdb")
"1.0.0"
"""