Source code for eoreader.stac.stac_utils
# -*- coding: utf-8 -*-
# Copyright 2024, SERTIT-ICube - France, https://sertit.unistra.fr/
# This file is part of eoreader project
# https://github.com/sertit/eoreader
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
STAC utils
"""
import os
from datetime import datetime
from pprint import pformat
from typing import Any
import geopandas as gpd
from sertit.vectors import WGS84
from shapely.geometry import mapping
from eoreader.stac._stac_keywords import DESCRIPTION, GSD, TITLE
[docs]def fill_common_mtd(asset: Any, prod, **kwargs) -> None:
"""
Fill common metadata of STAC assets or items
Args:
asset (pystac.Asset, pystac.Item): Asset or item
prod (Product): EOReader product
**kwargs: Additional arguments
"""
# Basics
asset.common_metadata.title = kwargs.get(TITLE, "No title available")
asset.common_metadata.description = kwargs.get(DESCRIPTION, "")
if len(asset.common_metadata.description) < 1:
asset.common_metadata.description = "No description available"
# Date and Time
asset.common_metadata.created = datetime.utcnow()
asset.common_metadata.updated = datetime.utcnow()
# Licensing
# asset.common_metadata.license = None # Collection level if possible
# Provider
# asset.common_metadata.providers = None # Collection level if possible
# Date and Time Range
asset.common_metadata.start_datetime = prod.datetime
asset.common_metadata.end_datetime = prod.datetime # TODO
# Instrument
asset.common_metadata.platform = None # TODO
asset.common_metadata.instruments = [
prod.instrument if isinstance(prod.instrument, str) else prod.instrument.value
]
asset.common_metadata.constellation = prod.constellation.value.lower()
asset.common_metadata.mission = None
asset.common_metadata.gsd = kwargs.get(GSD)
[docs]def gdf_to_geometry(gdf: gpd.GeoDataFrame) -> dict:
"""
Convert a geodataframe to a STAC geometry
Args:
gdf (gpd.GeoDataFrame): Geodataframe to convert to geometry
Returns:
dict: STAC Geometry
"""
return mapping(gdf.geometry.iat[0])
[docs]def gdf_to_bbox(gdf: gpd.GeoDataFrame) -> list:
"""
Convert a geodataframe to a STAC bbox
Args:
gdf (gpd.GeoDataFrame): Geodataframe to convert to bbox
Returns:
dict: STAC bbox
"""
# Convert from numpy dtype (which are not JSON serializable) to standard dtype
return [float(val) for val in gdf.bounds.values[0]]
[docs]def gdf_to_centroid(gdf: gpd.GeoDataFrame) -> dict:
"""
Convert a geodataframe to a STAC centroid
Args:
gdf (gpd.GeoDataFrame): Geodataframe to convert to centroid
Returns:
dict: STAC centroid
"""
centroid = gdf.centroid.to_crs(WGS84).iat[0]
# Convert from numpy dtype (which are not JSON serializable) to standard dtype
return {"lat": float(centroid.y), "lon": float(centroid.x)}
[docs]def repr_multiline_str(to_str: Any, nof_tabs: int) -> str:
"""
Format multiline string repr of a list, dict... by adding tabs at the beginning of every new line.
Begins with a new line.
Args:
to_str (Any): List, dict... to convert to tabbed string
nof_tabs (int): Number of tabs to put at the beginning of the lines
Returns:
"""
nof_tabs_str = "\t" * nof_tabs
return f"\n{nof_tabs_str}" + f"{nof_tabs_str}".join(
pformat(to_str).splitlines(True)
)
[docs]def to_float(val: Any) -> float:
"""
Convert from numpy dtype (which are not JSON serializable) to standard dtype
Args:
val (Any): Value in numpy dtype to be converted to float
Returns:
float: Value converted
"""
try:
# Convert from numpy dtype (which are not JSON serializable) to standard dtype
return float(val)
except TypeError:
return val
