"""
Implementation of base Sentinel Hub interfaces
"""
from __future__ import annotations
from abc import ABCMeta, abstractmethod
from typing import Any
from ..base import DataRequest
from ..constants import MimeType, MosaickingOrder, RequestType, ResamplingType
from ..data_collections import DataCollection, OrbitDirection
from ..download import DownloadRequest
from ..geometry import BBox, Geometry
from ..time_utils import RawTimeIntervalType, parse_time_interval, serialize_time
from .utils import _update_other_args
[docs]class SentinelHubBaseApiRequest(DataRequest, metaclass=ABCMeta):
"""A base class for Sentinel Hub interfaces"""
_SERVICE_ENDPOINT = ""
payload: dict[str, Any] = {} # noqa: RUF012
@property
@abstractmethod
def mime_type(self) -> MimeType:
"""The mime type of the request."""
[docs] def create_request(self) -> None:
"""Prepares a download request"""
headers = {"content-type": MimeType.JSON.get_string(), "accept": self.mime_type.get_string()}
base_url = self._get_base_url()
self.download_list = [
DownloadRequest(
request_type=RequestType.POST,
url=f"{base_url}/api/v1/{self._SERVICE_ENDPOINT}",
post_values=self.payload,
data_folder=self.data_folder,
save_response=bool(self.data_folder),
data_type=self.mime_type,
headers=headers,
use_session=True,
)
]
[docs] @staticmethod
def bounds(
bbox: BBox | None = None, geometry: Geometry | None = None, other_args: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Generate a `bound` part of the API request
:param bbox: Bounding box describing the area of interest.
:param geometry: Geometry describing the area of interest.
:param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
by it.
"""
if bbox is None and geometry is None:
raise ValueError("'bbox' and/or 'geometry' have to be provided.")
if bbox and not isinstance(bbox, BBox):
raise ValueError("'bbox' should be an instance of sentinelhub.BBox")
if geometry and not isinstance(geometry, Geometry):
raise ValueError("'geometry' should be an instance of sentinelhub.Geometry")
if bbox and geometry and bbox.crs != geometry.crs:
raise ValueError("bbox and geometry should be in the same CRS")
crs = bbox.crs if bbox else geometry.crs # type: ignore[union-attr]
request_bounds: dict[str, Any] = {"properties": {"crs": crs.opengis_string}}
if bbox:
request_bounds["bbox"] = list(bbox)
if geometry:
request_bounds["geometry"] = geometry.get_geojson(with_crs=False)
if other_args:
_update_other_args(request_bounds, other_args)
return request_bounds
def _get_base_url(self) -> str:
"""It decides which base URL to use. Restrictions from data collection definitions overrule the
settings from config object. In case different collections have different restrictions then
`SHConfig.sh_base_url` breaks the tie in case it matches one of the data collection URLs.
"""
data_collection_urls = tuple(
{
input_data_dict.service_url.rstrip("/")
for input_data_dict in self.payload["input"]["data"]
if isinstance(input_data_dict, InputDataDict) and input_data_dict.service_url is not None
}
)
config_base_url = self.config.sh_base_url.rstrip("/")
if not data_collection_urls:
return config_base_url
if len(data_collection_urls) == 1:
return data_collection_urls[0]
if config_base_url in data_collection_urls:
return config_base_url
raise ValueError(
f"Given data collections are restricted to different services: {data_collection_urls}\n"
"Configuration parameter sh_base_url cannot break the tie because it is set to a different"
f"service: {config_base_url}"
)
def _get_data_filters(
data_collection: DataCollection,
time_interval: RawTimeIntervalType | None,
maxcc: float | None,
mosaicking_order: MosaickingOrder | None,
) -> dict[str, Any]:
"""Builds a dictionary of data filters for Process API"""
data_filter: dict[str, Any] = {}
if time_interval:
start_time, end_time = serialize_time(parse_time_interval(time_interval, allow_undefined=True), use_tz=True)
data_filter["timeRange"] = {"from": start_time, "to": end_time}
if maxcc is not None:
if maxcc < 0 or maxcc > 1:
raise ValueError("maxcc should be a float on an interval [0, 1]")
data_filter["maxCloudCoverage"] = int(maxcc * 100)
if mosaicking_order:
data_filter["mosaickingOrder"] = MosaickingOrder(mosaicking_order).value
return {**data_filter, **_get_data_collection_filters(data_collection)}
def _get_data_collection_filters(data_collection: DataCollection) -> dict[str, Any]:
"""Builds a dictionary of filters for Process API from a data collection definition"""
filters: dict[str, Any] = {}
if data_collection.swath_mode:
filters["acquisitionMode"] = data_collection.swath_mode.upper()
if data_collection.polarization:
filters["polarization"] = data_collection.polarization.upper()
if data_collection.resolution:
filters["resolution"] = data_collection.resolution.upper()
if data_collection.orbit_direction and data_collection.orbit_direction.upper() != OrbitDirection.BOTH:
filters["orbitDirection"] = data_collection.orbit_direction.upper()
if data_collection.timeliness:
filters["timeliness"] = data_collection.timeliness
if data_collection.dem_instance:
filters["demInstance"] = data_collection.dem_instance
return filters
def _get_processing_params(upsampling: ResamplingType | None, downsampling: ResamplingType | None) -> dict[str, Any]:
"""Builds a dictionary of processing parameters for Process API"""
processing_params: dict[str, Any] = {}
if upsampling:
processing_params["upsampling"] = ResamplingType(upsampling).value
if downsampling:
processing_params["downsampling"] = ResamplingType(downsampling).value
return processing_params