from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date
from typing import Dict, List, Union
import dinamis_sdk
import planetary_computer
from pystac_client import Client, CollectionClient
from simplestac.utils import ItemCollection
from spectral_indices.roi.bbox import BoundingBox
from spectral_indices.sources.masks import Masks
from spectral_indices.sources.source_masks import SENTINEL2L2A_THEIA_MASKS
[docs]
@dataclass
class DataSource:
"""DataSource wraps all informations for a source of remote sensing data accessible trhough STAC catalog.
Args:
name (str): Name of the data source.
catalog (str): URL of data source STAC catalog.
collection_name (str): Name of the collection in the STAC catalog.
adaptater (BandAdaptater): Band adaptater to map data source assets to gloabl Band definition.
modifier (callable): Function to sign assets.
sensor (str): Name of the sensor in index database.
bands (``List[str]``): Available assets from this STAC collection.
"""
name: str = field(default_factory=str)
catalog: str = field(default_factory=str)
collection_name: str = field(default_factory=str)
modifier: callable = None
masks: Dict[str, str] = field(default=Masks({}))
bands: List[str] = field(default_factory=list)
sensor_name: str = field(default_factory=str)
[docs]
def fetch_collection(
self,
bbox: BoundingBox,
timestamps: Union[Union[str, date], List[Union[str, date]]],
) -> ItemCollection:
"""Fetch a collection of items from a STAC catalog.
Args:
bbox (``BoundingBox``): BoundingBox that represent extent of the items to query.
timestamps (``Union[Union[str, date],List[Union[str,date]]]``): Timestamps to query items from.
Returns:
``ItemCollection``:
- Return ItemCollection (simplestac extended) according to query.
"""
client = Client.open(url=self.catalog, modifier=self.modifier)
query = client.search(
collections=[self.collection_name],
intersects=bbox.transform("EPSG:4326"),
datetime=timestamps,
)
return ItemCollection(query.item_collection())
@property
def collection(self) -> CollectionClient:
"""Return whole collection information as Collection (pystac)."""
client = Client.open(url=self.catalog, modifier=self.modifier)
return client.get_collection(self.collection_name)
[docs]
class DataSourceFactory:
"""Registry to store Data sources and make it accessible through CLI."""
_sources: Dict[str, DataSource] = {}
[docs]
@classmethod
def register(cls, builder: DataSource) -> DataSource:
"""Register Data source. This can be used as a decorator.
Args:
builder (``DataSource``): Data source builder.
Returns:
``DataSource``:
- Return class builder.
"""
source: DataSource = builder()
cls._sources[source.name] = builder
return builder
[docs]
@classmethod
def get(cls, name: str) -> DataSource:
"""Gather a data source from registry using it's name."""
return cls._sources[name]()
[docs]
@DataSourceFactory.register
class Sentinel2L2ATheia:
"""Data source from Sentinel2L2A data from THEIA Montpellier STAC catalog."""
def __new__(self) -> Sentinel2L2ATheia:
catalog = "https://stacapi-cdos.apps.okd.crocc.meso.umontpellier.fr"
collection_name = "sentinel2-l2a-theia"
name = "sentinel2l2a_theia"
modifier = dinamis_sdk.sign_inplace
sensor_name = "sentinel2"
bands = ["B02", "B03", "B04", "B05", "B06", "B07", "B08", "B11", "B12", "B8A"]
masks = Masks(SENTINEL2L2A_THEIA_MASKS)
return DataSource(
name,
catalog,
collection_name,
modifier,
sensor_name=sensor_name,
bands=bands,
masks=masks,
)
[docs]
@DataSourceFactory.register
class MPlanetarySentinel2L2A:
"""Sentinel2L2A from Microsoft Planetary STAC catalog."""
def __new__(self) -> MPlanetarySentinel2L2A:
catalog = "https://planetarycomputer.microsoft.com/api/stac/v1"
collection_name = "sentinel-2-l2a"
name = "sentinel2l2a_microsoft_planetary"
modifier = planetary_computer.sign_inplace
sensor_name = "sentinel2"
bands = [
"B02",
"B03",
"B04",
"B05",
"B06",
"B07",
"B08",
"B8A",
"B09",
"B11",
"B12",
]
return DataSource(
name,
catalog,
collection_name,
modifier,
sensor_name=sensor_name,
bands=bands,
)