Skip to content
Snippets Groups Projects
raster2stac.py 42.1 KiB
Newer Older
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                    float(self.data.coords[self.X_DIM].max()),
                ],
                "reference_system": int((self.data.rio.crs.to_string()).split(":")[1]),
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
            },
            self.Y_DIM: {
                "axis": "y",
                "type": "spatial",
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                "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]),
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
            },
            self.T_DIM: {
                "type": "temporal",
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                "extent": [
                    str(self.data[self.T_DIM].min().values),
                    str(self.data[self.T_DIM].max().values),
                ],
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
            },
            self.B_DIM: {
                "type": "bands",
                "values": list(self.data[self.B_DIM].values),
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
        }

        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,
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                assets=collection_assets,
            )
        else:
            self.stac_collection = pystac.collection.Collection(
                id=self.collection_id,
                description=self.description,
                extent=pystac.Extent(spatial=s_ext, temporal=t_ext),
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                extra_fields=extra_fields,
        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,
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                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(
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
            pystac.Link(
                pystac.RelType.SELF,
                f"{self.fix_path_slash(self.collection_url)}{self.collection_id}",
                media_type=pystac.MediaType.JSON,
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
        # self.stac_collection.remove_links(rel=pystac.RelType.ROOT)
        self.stac_collection.add_link(
            pystac.Link(
                pystac.RelType.ROOT,
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                self.get_root_url(
                    f"{self.fix_path_slash(self.collection_url)}{self.collection_id}/items"
                ),
                media_type=pystac.MediaType.JSON,
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed

        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()
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
        # in order to solve the double "root" link bug/issue
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
        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):
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
                if (
                    link["rel"] == "root"
                    and link["href"] == links_dict[self_idx]["href"]
                    and link["type"] == links_dict[self_idx]["type"]
                ):
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
                    del links_dict[idx]
                    break
        if self.links is not None:
            stac_collection_dict["links"] = stac_collection_dict["links"] + self.links

Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
        # if self.output_format == "json_full":
        # stac_collection_dict["features"] = item_list  # Replace the "features" field with the list of items

        json_str = json.dumps(stac_collection_dict, indent=4)
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
        # printing metadata.json test output file
Claus Michele's avatar
Claus Michele committed
        output_path = Path(self.output_folder) / Path(self.output_file)
        with open(output_path, "w+") as metadata:
            metadata.write(json_str)
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
        if self.s3_upload:
Rufai Omowunmi Balogun's avatar
Rufai Omowunmi Balogun committed
            # 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)}'
            )
Lorenzo.Mercurio's avatar
Lorenzo.Mercurio committed
            self.upload_s3(output_path)