1""" 2Implementation of Sentinel Hub Process API interface 3""" 4from .constants import MimeType, RequestType 5from .download import DownloadRequest 6from .data_collections import OrbitDirection 7from .data_request import DataRequest 8from .geometry import Geometry, BBox 9from .sh_utils import _update_other_args 10from .time_utils import parse_time_interval, serialize_time 11 12 13class SentinelHubBaseApiRequest(DataRequest): 14 """ A base class for Sentinel Hub interfaces 15 """ 16 _SERVICE_ENDPOINT = '' 17 18 def create_request(self): 19 """ Prepares a download request 20 """ 21 headers = {'content-type': MimeType.JSON.get_string(), 'accept': self.mime_type.get_string()} 22 base_url = self._get_base_url() 23 self.download_list = [DownloadRequest( 24 request_type=RequestType.POST, 25 url=f'{base_url}/api/v1/{self._SERVICE_ENDPOINT}', 26 post_values=self.payload, 27 data_folder=self.data_folder, 28 save_response=bool(self.data_folder), 29 data_type=self.mime_type, 30 headers=headers, 31 use_session=True 32 )] 33 34 @staticmethod 35 def input_data(data_collection, *, identifier=None, time_interval=None, maxcc=None, mosaicking_order=None, 36 upsampling=None, downsampling=None, other_args=None): 37 """ Generate the `input data` part of the request body 38 39 :param data_collection: One of supported Process API data collections. 40 :type data_collection: DataCollection 41 :param identifier: A collection identifier that can be referred to in the evalscript. Parameter is referenced 42 as `"id"` in service documentation. To learn more check 43 `data fusion documentation <https://docs.sentinel-hub.com/api/latest/data/data-fusion>`__. 44 :type identifier: str or None 45 :param time_interval: A time interval with start and end date of the form YYYY-MM-DDThh:mm:ss or YYYY-MM-DD or 46 a datetime object 47 :type time_interval: (str, str) or (datetime, datetime) 48 :param maxcc: Maximum accepted cloud coverage of an image. Float between 0.0 and 1.0. Default is 1.0. 49 :type maxcc: float or None 50 :param mosaicking_order: Mosaicking order, which has to be either 'mostRecent', 'leastRecent' or 'leastCC'. 51 :type mosaicking_order: str or None 52 :param upsampling: A type of upsampling to apply on data 53 :type upsampling: str 54 :param downsampling: A type of downsampling to apply on data 55 :type downsampling: str 56 :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated 57 by it. 58 :type other_args: dict 59 :return: A dictionary-like object that also contains additional attributes 60 :rtype: InputDataDict 61 """ 62 input_data_dict = { 63 'type': data_collection.api_id, 64 } 65 if identifier: 66 input_data_dict['id'] = identifier 67 68 data_filters = _get_data_filters(data_collection, time_interval, maxcc, mosaicking_order) 69 if data_filters: 70 input_data_dict['dataFilter'] = data_filters 71 72 processing_params = _get_processing_params(upsampling, downsampling) 73 if processing_params: 74 input_data_dict['processing'] = processing_params 75 76 if other_args: 77 _update_other_args(input_data_dict, other_args) 78 79 return InputDataDict(input_data_dict, service_url=data_collection.service_url) 80 81 @staticmethod 82 def bounds(bbox=None, geometry=None, other_args=None): 83 """ Generate a `bound` part of the API request 84 85 :param bbox: Bounding box describing the area of interest. 86 :type bbox: sentinelhub.BBox 87 :param geometry: Geometry describing the area of interest. 88 :type geometry: sentinelhub.Geometry 89 :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated 90 by it. 91 :type other_args: dict 92 """ 93 if bbox is None and geometry is None: 94 raise ValueError("'bbox' and/or 'geometry' have to be provided.") 95 96 if bbox and not isinstance(bbox, BBox): 97 raise ValueError("'bbox' should be an instance of sentinelhub.BBox") 98 99 if geometry and not isinstance(geometry, Geometry): 100 raise ValueError("'geometry' should be an instance of sentinelhub.Geometry") 101 102 if bbox and geometry and bbox.crs != geometry.crs: 103 raise ValueError('bbox and geometry should be in the same CRS') 104 105 crs = bbox.crs if bbox else geometry.crs 106 107 request_bounds = { 108 'properties': { 109 'crs': crs.opengis_string 110 } 111 } 112 113 if bbox: 114 request_bounds['bbox'] = list(bbox) 115 116 if geometry: 117 request_bounds['geometry'] = geometry.get_geojson(with_crs=False) 118 119 if other_args: 120 _update_other_args(request_bounds, other_args) 121 122 return request_bounds 123 124 def _get_base_url(self): 125 """ It decides which base URL to use. Restrictions from data collection definitions overrule the 126 settings from config object. In case different collections have different restrictions then 127 `SHConfig.sh_base_url` breaks the tie in case it matches one of the data collection URLs. 128 """ 129 data_collection_urls = tuple({ 130 input_data_dict.service_url.rstrip('/') for input_data_dict in self.payload['input']['data'] 131 if isinstance(input_data_dict, InputDataDict) and input_data_dict.service_url is not None 132 }) 133 config_base_url = self.config.sh_base_url.rstrip('/') 134 135 if not data_collection_urls: 136 return config_base_url 137 138 if len(data_collection_urls) == 1: 139 return data_collection_urls[0] 140 141 if config_base_url in data_collection_urls: 142 return config_base_url 143 144 raise ValueError(f'Given data collections are restricted to different services: {data_collection_urls}\n' 145 'Configuration parameter sh_base_url cannot break the tie because it is set to a different' 146 f'service: {config_base_url}') 147 148 149class InputDataDict(dict): 150 """ An input data dictionary which also holds additional attributes 151 """ 152 def __init__(self, input_data_dict, *, service_url=None): 153 """ 154 :param input_data_dict: A normal dictionary with input parameters 155 :type input_data_dict: dict 156 :param service_url: A service URL defined by a data collection 157 :type service_url: str 158 """ 159 super().__init__(input_data_dict) 160 self.service_url = service_url 161 162 def __repr__(self): 163 """ Modified dictionary representation that also shows additional attributes 164 """ 165 normal_dict_repr = super().__repr__() 166 return f'{self.__class__.__name__}({normal_dict_repr}, service_url={self.service_url})' 167 168 169def _get_data_filters(data_collection, time_interval, maxcc, mosaicking_order): 170 """ Builds a dictionary of data filters for Process API 171 """ 172 data_filter = {} 173 174 if time_interval: 175 start_time, end_time = serialize_time(parse_time_interval(time_interval, allow_undefined=True), use_tz=True) 176 data_filter['timeRange'] = {'from': start_time, 'to': end_time} 177 178 if maxcc is not None: 179 if not isinstance(maxcc, float) and (maxcc < 0 or maxcc > 1): 180 raise ValueError('maxcc should be a float on an interval [0, 1]') 181 182 data_filter['maxCloudCoverage'] = int(maxcc * 100) 183 184 if mosaicking_order: 185 mosaic_order_params = ['mostRecent', 'leastRecent', 'leastCC'] 186 187 if mosaicking_order not in mosaic_order_params: 188 raise ValueError(f'{mosaicking_order} is not a valid mosaickingOrder parameter, it should be one ' 189 f'of: {mosaic_order_params}') 190 191 data_filter['mosaickingOrder'] = mosaicking_order 192 193 return { 194 **data_filter, 195 **_get_data_collection_filters(data_collection) 196 } 197 198 199def _get_data_collection_filters(data_collection): 200 """ Builds a dictionary of filters for Process API from a data collection definition 201 """ 202 filters = {} 203 204 if data_collection.swath_mode: 205 filters['acquisitionMode'] = data_collection.swath_mode.upper() 206 207 if data_collection.polarization: 208 filters['polarization'] = data_collection.polarization.upper() 209 210 if data_collection.resolution: 211 filters['resolution'] = data_collection.resolution.upper() 212 213 if data_collection.orbit_direction and data_collection.orbit_direction.upper() != OrbitDirection.BOTH: 214 filters['orbitDirection'] = data_collection.orbit_direction.upper() 215 216 if data_collection.timeliness: 217 filters['timeliness'] = data_collection.timeliness 218 219 if data_collection.dem_instance: 220 filters['demInstance'] = data_collection.dem_instance 221 222 return filters 223 224 225def _get_processing_params(upsampling, downsampling): 226 """ Builds a dictionary of processing parameters for Process API 227 """ 228 processing_params = {} 229 230 if upsampling: 231 processing_params['upsampling'] = upsampling 232 233 if downsampling: 234 processing_params['downsampling'] = downsampling 235 236 return processing_params 237