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
|
"""
Script to download the examples from the stac-spec repository.
This is used when upgrading to a new version of STAC.
"""
import argparse
import json
import os
import tempfile
from subprocess import call
from typing import Any
from urllib.error import HTTPError
import pystac
from pystac.serialization import identify_stac_object
def remove_bad_collection(js: dict[str, Any]) -> dict[str, Any]:
links: list[dict[str, Any]] | None = js.get("links")
if links is not None:
filtered_links: list[dict[str, Any]] = []
for link in links:
rel = link.get("rel")
if rel is not None and rel == "collection":
href: str = link["href"]
try:
json.loads(pystac.StacIO.default().read_text(href))
filtered_links.append(link)
except (HTTPError, FileNotFoundError, json.decoder.JSONDecodeError):
print(f"===REMOVING UNREADABLE COLLECTION AT {href}")
else:
filtered_links.append(link)
js["links"] = filtered_links
return js
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Get examples from the stac-spec repo."
)
parser.add_argument(
"previous_version",
metavar="PREVIOUS_VERSION",
help="The previous STAC_VERSION that examples have already been pulled from.",
)
args = parser.parse_args()
stac_repo = "https://github.com/radiantearth/stac-spec"
stac_spec_tag = f"v{pystac.get_stac_version()}"
examples_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "examples"))
with tempfile.TemporaryDirectory() as tmp_dir:
call(
[
"git",
"clone",
"--depth",
"1",
"--branch",
stac_spec_tag,
stac_repo,
tmp_dir,
]
)
example_dirs: list[str] = []
for root, _, _ in os.walk(tmp_dir):
example_dirs.append(os.path.join(root))
example_csv_lines = set()
for example_dir in example_dirs:
for root, _, files in os.walk(example_dir):
for fname in files:
if fname.endswith(".json"):
path = os.path.join(root, fname)
with open(path) as f:
try:
js: dict[str, Any] = json.loads(f.read())
except json.decoder.JSONDecodeError:
# Account for bad examples that can't be parsed.
js = {}
example_version = js.get("stac_version")
if (
example_version is not None
and example_version > args.previous_version
):
relpath = "{}/{}".format(
pystac.get_stac_version(),
path.replace(f"{tmp_dir}/", ""),
)
target_path = os.path.join(examples_dir, relpath)
print(f"Creating example at {target_path}")
info = identify_stac_object(js)
# Handle the case where there are collection links that
# don't exist.
if info.object_type == pystac.STACObjectType.ITEM:
js = remove_bad_collection(js)
d = os.path.dirname(target_path)
if not os.path.isdir(d):
os.makedirs(d)
with open(target_path, "w") as f:
f.write(json.dumps(js, indent=4))
# Add info to the new example-info.csv lines
line_info: list[str] = [
relpath,
info.object_type,
example_version,
"|".join(info.extensions),
]
line = '"{}"'.format('","'.join(line_info))
example_csv_lines.add(line)
# Write the new example-info.csv lines into a temp file for inspection
with open(os.path.join(examples_dir, "examples-info-NEW.csv"), "w") as f:
txt = "\n".join(sorted(example_csv_lines))
f.write(txt)
|
