"""
Implementation of base Sentinel Hub interfaces
"""
from abc import ABCMeta, abstractmethod
from typing import Any, Dict, Optional
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] = {}
@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: Optional[BBox] = None, geometry: Optional[Geometry] = None, other_args: Optional[Dict[str, Any]] = 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: Optional[RawTimeIntervalType],
maxcc: Optional[float],
mosaicking_order: Optional[MosaickingOrder],
) -> 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: Optional[ResamplingType], downsampling: Optional[ResamplingType]
) -> 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