Newer
Older
"extent": [
float(self.data.coords[self.X_DIM].min()),
float(self.data.coords[self.X_DIM].max()),
],
"reference_system": int(
self.data.rio.crs.to_string().split(":")[1]
},
self.Y_DIM: {
"axis": "y",
"type": "spatial",
"extent": [
float(self.data.coords[self.Y_DIM].min()),
float(self.data.coords[self.Y_DIM].max()),
],
"reference_system": int(
self.data.rio.crs.to_string().split(":")[1]
"extent": [
str(self.data[self.T_DIM].min().values),
str(self.data[self.T_DIM].max().values),
],
},
self.B_DIM: {
"type": "bands",
"values": list(self.data[self.B_DIM].values),
}
extra_fields["cube:dimensions"] = cube_dimensons
if self.write_collection_assets:
self.stac_collection = pystac.collection.Collection(
id=self.collection_id,
description=self.description,
extent=pystac.Extent(spatial=s_ext, temporal=t_ext),
extra_fields=extra_fields,
)
else:
self.stac_collection = pystac.collection.Collection(
id=self.collection_id,
description=self.description,
extent=pystac.Extent(spatial=s_ext, temporal=t_ext),
self.stac_collection.add_link(
pystac.Link(
pystac.RelType.ITEMS,
f"{self.fix_path_slash(self.collection_url)}{self.collection_id}/items",
media_type=pystac.MediaType.JSON,
self.stac_collection.add_link(
pystac.Link(
pystac.RelType.PARENT,
self.get_root_url(
f"{self.fix_path_slash(self.collection_url)}{self.collection_id}/items"
),
media_type=pystac.MediaType.JSON,
self.stac_collection.add_link(
pystac.RelType.SELF,
f"{self.fix_path_slash(self.collection_url)}{self.collection_id}",
media_type=pystac.MediaType.JSON,
# self.stac_collection.remove_links(rel=pystac.RelType.ROOT)
self.stac_collection.add_link(
pystac.Link(
pystac.RelType.ROOT,
self.get_root_url(
f"{self.fix_path_slash(self.collection_url)}{self.collection_id}/items"
),
media_type=pystac.MediaType.JSON,
if self.license is not None:
self.stac_collection.license = self.license
# Create a single JSON file with all the items
stac_collection_dict = self.stac_collection.to_dict()
# in order to solve the double "root" link bug/issue
links_dict = stac_collection_dict["links"]
ctr_roots = 0
self_exists = False
self_idx = 0
for idx, link in enumerate(links_dict):
if link["rel"] == "root":
ctr_roots = ctr_roots + 1
if link["rel"] == "self":
self_exists = True
self_idx = idx
if ctr_roots == 2 and self_exists:
for idx, link in enumerate(links_dict):
if (
link["rel"] == "root"
and link["href"] == links_dict[self_idx]["href"]
and link["type"] == links_dict[self_idx]["type"]
):
if self.links is not None:
stac_collection_dict["links"] = stac_collection_dict["links"] + self.links
json_str = json.dumps(stac_collection_dict, indent=4)
output_path = Path(self.output_folder) / Path(self.output_file)
with open(output_path, "w+") as metadata:
# Uploading metadata JSON file to s3
_log.debug(
f'Uploading metatada JSON "{output_path}" to {self.fix_path_slash(self.bucket_file_prefix)}{os.path.basename(output_path)}'
)