Skip to content

Catalog

Bases: CatalogBase, VizTools

The Catalog class enables access to the UP42 catalog functionality (data archive search & ordering).

Use catalog:

catalog = up42.initialize_catalog()

This class also inherits functions from the CatalogBase class.

Source code in up42/catalog.py
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
class Catalog(CatalogBase, VizTools):
    """
    The Catalog class enables access to the UP42 catalog functionality (data archive search & ordering).

    Use catalog:
    ```python
    catalog = up42.initialize_catalog()
    ```

    This class also inherits functions from the [CatalogBase](catalogbase-reference.md) class.
    """

    def __init__(self, auth: Auth):
        self.auth = auth
        self.quicklooks = None
        self.type = "ARCHIVE"
        self.data_products: Union[None, dict] = None

    def __repr__(self):
        return f"Catalog(auth={self.auth})"

    def estimate_order(self, order_parameters: Union[dict, None], **kwargs) -> int:
        """
        Estimate the cost of an order.

        Args:
            order_parameters: A dictionary like {dataProduct: ..., "params": {"id": ..., "aoi": ...}}

        Returns:
            int: An estimated cost for the order in UP42 credits.

        Warning "Deprecated order parameters"
            The use of the 'scene' and 'geometry' parameters for the data estimation is deprecated. Please use the new
            order_parameters parameter as described above.
        """
        if "scene" in kwargs or "geometry" in kwargs:
            # Deprecated, to be removed, use order_parameters.
            message = (
                "The use of the 'scene' and 'geometry' parameters for the data estimation is deprecated. "
                "Please use the new 'order_parameters' parameter."
            )
            warnings.warn(message, DeprecationWarning, stacklevel=2)
        elif order_parameters is None:
            raise ValueError("Please provide the 'order_parameters' parameter!")
        return Order.estimate(self.auth, order_parameters)  # type: ignore

    @deprecation("construct_search_parameters", "0.25.0")
    def construct_parameters(self, **kwargs):  # pragma: no cover
        """Deprecated, see construct_search_parameters"""
        return self.construct_search_parameters(**kwargs)

    @staticmethod
    def construct_search_parameters(
        geometry: Union[FeatureCollection, Feature, dict, list, GeoDataFrame, Polygon],
        collections: List[str],
        start_date: str = "2020-01-01",
        end_date: str = "2020-01-30",
        usage_type: List[str] = None,
        limit: int = 10,
        max_cloudcover: Optional[int] = None,
        sortby: str = None,
        ascending: bool = None,
    ) -> dict:
        """
        Helps constructing the parameters dictionary required for the search.

        Args:
            geometry: The search geometry, default a Polygon. One of FeatureCollection, Feature,
                dict (geojson geometry), list (bounds coordinates), GeoDataFrame, shapely.Polygon, shapely.Point.
                All assume EPSG 4326!
            collections: The satellite sensor collections to search for, e.g. ["phr"] or ["phr", "spot"].
                Also see catalog.get_collections().
            start_date: Query period starting day, format "2020-01-01".
            end_date: Query period ending day, format "2020-01-01".
            usage_type: Optional. Filter for imagery that can be purchased & downloaded or also
                processed. ["DATA"] (can only be downloaded), ["ANALYTICS"] (can be downloaded
                or used directly with a processing algorithm), ["DATA", "ANALYTICS"]
                (can be any combination). The filter is inclusive, using ["DATA"] can
                also result in results with ["DATA", "ANALYTICS"].
            limit: The maximum number of search results to return (1-max.500).
            max_cloudcover: Optional. Maximum cloud coverage percent - e.g. 100 will return all scenes,
                8.4 will return all scenes with 8.4 or less cloud coverage.
            sortby: (deprecated)
            ascending: (deprecated)
        Returns:
            The constructed parameters dictionary.
        """

        if sortby is not None or ascending is not None:
            logger.info(
                "sortby is deprecated, currently only sorting output by creation date."
            )

        time_period = (
            f"{format_time(start_date)}/{format_time(end_date, set_end_of_day=True)}"
        )
        aoi_fc = any_vector_to_fc(
            vector=geometry,
        )
        aoi_geometry = fc_to_query_geometry(fc=aoi_fc, geometry_operation="intersects")

        query_filters: Dict[Any, Any] = {}
        if max_cloudcover is not None:
            query_filters["cloudCoverage"] = {"lte": max_cloudcover}  # type: ignore

        if usage_type is not None:
            if usage_type == ["DATA"]:
                query_filters["up42:usageType"] = {"in": ["DATA"]}
            elif usage_type == ["ANALYTICS"]:
                query_filters["up42:usageType"] = {"in": ["ANALYTICS"]}
            elif usage_type == ["DATA", "ANALYTICS"]:
                query_filters["up42:usageType"] = {"in": ["DATA", "ANALYTICS"]}
            else:
                raise ValueError("Select correct `usage_type`")

        search_parameters = {
            "datetime": time_period,
            "intersects": aoi_geometry,
            "limit": limit,
            "collections": collections,
            "query": query_filters,
        }

        return search_parameters

    def search(
        self, search_parameters: dict, as_dataframe: bool = True
    ) -> Union[GeoDataFrame, dict]:
        """
        Searches the catalog for the the search parameters and returns the metadata of
        the matching scenes.

        Args:
            search_parameters: The catalog search parameters, see example.
            as_dataframe: return type, GeoDataFrame if True (default), FeatureCollection if False.

        Returns:
            The search results as a GeoDataFrame, optionally as JSON dict.

        Example:
            ```python
                search_parameters={
                    "datetime": "2019-01-01T00:00:00Z/2019-01-15T23:59:59Z",
                    "collections": ["phr"],
                    "intersects": {
                        "type": "Polygon",
                        "coordinates": [[[13.32113746,52.73971768],[13.15981158,52.2092959],
                        [13.62204483,52.15632025],[13.78859517,52.68655119],[13.32113746,
                        52.73971768]]]},
                    "limit": 10,
                    "sortby": [{"field" : "properties.acquisitionDate", "direction" : "asc"}]
                    }
            ```
        """
        logger.info(f"Searching catalog with search_parameters: {search_parameters}")

        # The API request would fail with a limit above 500, thus 500 is forced in the initial
        # request but additional results are handled below via pagination.
        try:
            max_limit = search_parameters["limit"]
        except KeyError:
            logger.info("No `limit` parameter in search_parameters, using default 500.")
            max_limit = 500

        if max_limit > 500:
            search_parameters["limit"] = 500

        # UP42 API can query multiple collections of the same host at once.
        if self.data_products is None:
            self.data_products = self.get_data_products(basic=True)  # type: ignore
        hosts = [
            v["host"]
            for v in self.data_products.values()  # type: ignore
            if v["collection"] in search_parameters["collections"]
        ]
        if not hosts:
            raise ValueError(
                f"Selected collections {search_parameters['collections']} are not valid. See "
                f"catalog.get_collections."
            )
        if len(set(hosts)) > 1:
            raise ValueError(
                "Only collections with the same host can be searched at the same time. Please adjust the "
                "collections in the search_parameters!"
            )
        host = hosts[0]

        url = f"{self.auth._endpoint()}/catalog/hosts/{host}/stac/search"
        response_json: dict = self.auth._request("POST", url, search_parameters)
        features = response_json["features"]

        # Search results with more than 500 items are given as 50-per-page additional pages.
        while len(features) < max_limit:
            pagination_exhausted = len(response_json["links"]) == 1
            if pagination_exhausted:
                break
            next_page_url = response_json["links"][1]["href"]
            response_json = self.auth._request("POST", next_page_url, search_parameters)
            features += response_json["features"]

        features = features[:max_limit]
        if not features:
            df = GeoDataFrame(columns=["geometry"], geometry="geometry")
        else:
            df = GeoDataFrame.from_features(
                FeatureCollection(features=features), crs="EPSG:4326"
            )

        logger.info(f"{df.shape[0]} results returned.")
        if as_dataframe:
            return df
        else:
            return df.__geo_interface__

    def construct_order_parameters(
        self,
        data_product_id: str,
        image_id: str,
        aoi: Union[
            dict,
            Feature,
            FeatureCollection,
            list,
            GeoDataFrame,
            Polygon,
        ] = None,
        tags: Optional[List[str]] = None,
    ):
        """
        Helps constructing the parameters dictionary required for the catalog order. Some collections have
        additional parameters that are added to the output dictionary with value None. The potential values to
        select from are given in the logs, for more detail on the parameter use `catalog.get_data_product_schema()`.

        Args:
            data_product_id: Id of the desired UP42 data product, see `catalog.get_data_products`
            image_id: The id of the desired image (from search results)
            aoi: The geometry of the order, one of dict, Feature, FeatureCollection,
                list, GeoDataFrame, Polygon. Optional for "full-image products".
            tags: A list of tags that categorize the order.
        Returns:
            The order parameters dictionary.

        Example:
            ```python
            order_parameters = catalog.construct_order_parameters(
                data_product_id='647780db-5a06-4b61-b525-577a8b68bb54',
                image_id='6434e7af-2d41-4ded-a789-fb1b2447ac92',
                aoi={'type': 'Polygon',
                'coordinates': (((13.375966, 52.515068),
                  (13.375966, 52.516639),
                  (13.378314, 52.516639),
                  (13.378314, 52.515068),
                  (13.375966, 52.515068)),)})
            ```
        """
        order_parameters = {
            "dataProduct": data_product_id,
            "params": {"id": image_id},
        }
        if tags is not None:
            order_parameters["tags"] = tags
        logger.info(
            "See `catalog.get_data_product_schema(data_product_id)` for more detail on the parameter options."
        )
        schema = self.get_data_product_schema(data_product_id)
        order_parameters = autocomplete_order_parameters(order_parameters, schema)

        # Some catalog orders, e.g. Capella don't require AOI (full image order)
        # Handled on API level, don't manipulate in SDK, providers might accept geometries in the future.
        if aoi is not None:
            aoi = any_vector_to_fc(vector=aoi)
            aoi = fc_to_query_geometry(fc=aoi, geometry_operation="intersects")
            order_parameters["params"]["aoi"] = aoi  # type: ignore

        return order_parameters

    def download_quicklooks(
        self,
        image_ids: List[str],
        collection: str,
        output_directory: Union[str, Path, None] = None,
    ) -> List[str]:
        """
        Gets the quicklooks of scenes from a single sensor. After download, can
        be plotted via catalog.map_quicklooks() or catalog.plot_quicklooks().

        Args:
            image_ids: List of provider image_ids e.g. ["6dffb8be-c2ab-46e3-9c1c-6958a54e4527"].
                Access the search results id column via `list(search_results.id)`.
            collection: The data collection corresponding to the image ids.
            output_directory: The file output directory, defaults to the current working
                directory.

        Returns:
            List of quicklook image output file paths.
        """
        if self.data_products is None:
            self.data_products = self.get_data_products(basic=True)  # type: ignore
        host = [
            v["host"]
            for v in self.data_products.values()  # type: ignore
            if v["collection"] == collection
        ]
        if not host:
            raise ValueError(
                f"Selected collections {collection} is not valid. See catalog.get_collections."
            )
        host = host[0]
        logger.info(f"Downloading quicklooks from provider {host}.")

        if output_directory is None:
            output_directory = Path.cwd() / f"project_{self.auth.project_id}/catalog"
        else:
            output_directory = Path(output_directory)
        output_directory.mkdir(parents=True, exist_ok=True)
        logger.info(f"Download directory: {str(output_directory)}")

        if isinstance(image_ids, str):
            image_ids = [image_ids]

        out_paths: List[str] = []
        for image_id in tqdm(image_ids):
            try:
                url = (
                    f"{self.auth._endpoint()}/catalog/{host}/image/{image_id}/quicklook"
                )
                response = self.auth._request(
                    request_type="GET", url=url, return_text=False
                )
                out_path = output_directory / f"quicklook_{image_id}.jpg"
                out_paths.append(str(out_path))
                with open(out_path, "wb") as dst:
                    for chunk in response:
                        dst.write(chunk)
            except ValueError:
                logger.warning(
                    f"Image with id {image_id} does not have quicklook available. Skipping ..."
                )

        self.quicklooks = out_paths  # pylint: disable=attribute-defined-outside-init
        return out_paths

construct_order_parameters(data_product_id, image_id, aoi=None, tags=None)

Helps constructing the parameters dictionary required for the catalog order. Some collections have additional parameters that are added to the output dictionary with value None. The potential values to select from are given in the logs, for more detail on the parameter use catalog.get_data_product_schema().

Parameters:

Name Type Description Default
data_product_id str

Id of the desired UP42 data product, see catalog.get_data_products

required
image_id str

The id of the desired image (from search results)

required
aoi Union[dict, Feature, FeatureCollection, list, GeoDataFrame, Polygon]

The geometry of the order, one of dict, Feature, FeatureCollection, list, GeoDataFrame, Polygon. Optional for "full-image products".

None
tags Optional[List[str]]

A list of tags that categorize the order.

None
Example
order_parameters = catalog.construct_order_parameters(
    data_product_id='647780db-5a06-4b61-b525-577a8b68bb54',
    image_id='6434e7af-2d41-4ded-a789-fb1b2447ac92',
    aoi={'type': 'Polygon',
    'coordinates': (((13.375966, 52.515068),
      (13.375966, 52.516639),
      (13.378314, 52.516639),
      (13.378314, 52.515068),
      (13.375966, 52.515068)),)})
Source code in up42/catalog.py
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
def construct_order_parameters(
    self,
    data_product_id: str,
    image_id: str,
    aoi: Union[
        dict,
        Feature,
        FeatureCollection,
        list,
        GeoDataFrame,
        Polygon,
    ] = None,
    tags: Optional[List[str]] = None,
):
    """
    Helps constructing the parameters dictionary required for the catalog order. Some collections have
    additional parameters that are added to the output dictionary with value None. The potential values to
    select from are given in the logs, for more detail on the parameter use `catalog.get_data_product_schema()`.

    Args:
        data_product_id: Id of the desired UP42 data product, see `catalog.get_data_products`
        image_id: The id of the desired image (from search results)
        aoi: The geometry of the order, one of dict, Feature, FeatureCollection,
            list, GeoDataFrame, Polygon. Optional for "full-image products".
        tags: A list of tags that categorize the order.
    Returns:
        The order parameters dictionary.

    Example:
        ```python
        order_parameters = catalog.construct_order_parameters(
            data_product_id='647780db-5a06-4b61-b525-577a8b68bb54',
            image_id='6434e7af-2d41-4ded-a789-fb1b2447ac92',
            aoi={'type': 'Polygon',
            'coordinates': (((13.375966, 52.515068),
              (13.375966, 52.516639),
              (13.378314, 52.516639),
              (13.378314, 52.515068),
              (13.375966, 52.515068)),)})
        ```
    """
    order_parameters = {
        "dataProduct": data_product_id,
        "params": {"id": image_id},
    }
    if tags is not None:
        order_parameters["tags"] = tags
    logger.info(
        "See `catalog.get_data_product_schema(data_product_id)` for more detail on the parameter options."
    )
    schema = self.get_data_product_schema(data_product_id)
    order_parameters = autocomplete_order_parameters(order_parameters, schema)

    # Some catalog orders, e.g. Capella don't require AOI (full image order)
    # Handled on API level, don't manipulate in SDK, providers might accept geometries in the future.
    if aoi is not None:
        aoi = any_vector_to_fc(vector=aoi)
        aoi = fc_to_query_geometry(fc=aoi, geometry_operation="intersects")
        order_parameters["params"]["aoi"] = aoi  # type: ignore

    return order_parameters

construct_parameters(**kwargs)

Deprecated, see construct_search_parameters

Source code in up42/catalog.py
202
203
204
205
@deprecation("construct_search_parameters", "0.25.0")
def construct_parameters(self, **kwargs):  # pragma: no cover
    """Deprecated, see construct_search_parameters"""
    return self.construct_search_parameters(**kwargs)

construct_search_parameters(geometry, collections, start_date='2020-01-01', end_date='2020-01-30', usage_type=None, limit=10, max_cloudcover=None, sortby=None, ascending=None) staticmethod

Helps constructing the parameters dictionary required for the search.

Parameters:

Name Type Description Default
geometry Union[FeatureCollection, Feature, dict, list, GeoDataFrame, Polygon]

The search geometry, default a Polygon. One of FeatureCollection, Feature, dict (geojson geometry), list (bounds coordinates), GeoDataFrame, shapely.Polygon, shapely.Point. All assume EPSG 4326!

required
collections List[str]

The satellite sensor collections to search for, e.g. ["phr"] or ["phr", "spot"]. Also see catalog.get_collections().

required
start_date str

Query period starting day, format "2020-01-01".

'2020-01-01'
end_date str

Query period ending day, format "2020-01-01".

'2020-01-30'
usage_type List[str]

Optional. Filter for imagery that can be purchased & downloaded or also processed. ["DATA"] (can only be downloaded), ["ANALYTICS"] (can be downloaded or used directly with a processing algorithm), ["DATA", "ANALYTICS"] (can be any combination). The filter is inclusive, using ["DATA"] can also result in results with ["DATA", "ANALYTICS"].

None
limit int

The maximum number of search results to return (1-max.500).

10
max_cloudcover Optional[int]

Optional. Maximum cloud coverage percent - e.g. 100 will return all scenes, 8.4 will return all scenes with 8.4 or less cloud coverage.

None
sortby str

(deprecated)

None
ascending bool

(deprecated)

None
Source code in up42/catalog.py
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@staticmethod
def construct_search_parameters(
    geometry: Union[FeatureCollection, Feature, dict, list, GeoDataFrame, Polygon],
    collections: List[str],
    start_date: str = "2020-01-01",
    end_date: str = "2020-01-30",
    usage_type: List[str] = None,
    limit: int = 10,
    max_cloudcover: Optional[int] = None,
    sortby: str = None,
    ascending: bool = None,
) -> dict:
    """
    Helps constructing the parameters dictionary required for the search.

    Args:
        geometry: The search geometry, default a Polygon. One of FeatureCollection, Feature,
            dict (geojson geometry), list (bounds coordinates), GeoDataFrame, shapely.Polygon, shapely.Point.
            All assume EPSG 4326!
        collections: The satellite sensor collections to search for, e.g. ["phr"] or ["phr", "spot"].
            Also see catalog.get_collections().
        start_date: Query period starting day, format "2020-01-01".
        end_date: Query period ending day, format "2020-01-01".
        usage_type: Optional. Filter for imagery that can be purchased & downloaded or also
            processed. ["DATA"] (can only be downloaded), ["ANALYTICS"] (can be downloaded
            or used directly with a processing algorithm), ["DATA", "ANALYTICS"]
            (can be any combination). The filter is inclusive, using ["DATA"] can
            also result in results with ["DATA", "ANALYTICS"].
        limit: The maximum number of search results to return (1-max.500).
        max_cloudcover: Optional. Maximum cloud coverage percent - e.g. 100 will return all scenes,
            8.4 will return all scenes with 8.4 or less cloud coverage.
        sortby: (deprecated)
        ascending: (deprecated)
    Returns:
        The constructed parameters dictionary.
    """

    if sortby is not None or ascending is not None:
        logger.info(
            "sortby is deprecated, currently only sorting output by creation date."
        )

    time_period = (
        f"{format_time(start_date)}/{format_time(end_date, set_end_of_day=True)}"
    )
    aoi_fc = any_vector_to_fc(
        vector=geometry,
    )
    aoi_geometry = fc_to_query_geometry(fc=aoi_fc, geometry_operation="intersects")

    query_filters: Dict[Any, Any] = {}
    if max_cloudcover is not None:
        query_filters["cloudCoverage"] = {"lte": max_cloudcover}  # type: ignore

    if usage_type is not None:
        if usage_type == ["DATA"]:
            query_filters["up42:usageType"] = {"in": ["DATA"]}
        elif usage_type == ["ANALYTICS"]:
            query_filters["up42:usageType"] = {"in": ["ANALYTICS"]}
        elif usage_type == ["DATA", "ANALYTICS"]:
            query_filters["up42:usageType"] = {"in": ["DATA", "ANALYTICS"]}
        else:
            raise ValueError("Select correct `usage_type`")

    search_parameters = {
        "datetime": time_period,
        "intersects": aoi_geometry,
        "limit": limit,
        "collections": collections,
        "query": query_filters,
    }

    return search_parameters

download_quicklooks(image_ids, collection, output_directory=None)

Gets the quicklooks of scenes from a single sensor. After download, can be plotted via catalog.map_quicklooks() or catalog.plot_quicklooks().

Parameters:

Name Type Description Default
image_ids List[str]

List of provider image_ids e.g. ["6dffb8be-c2ab-46e3-9c1c-6958a54e4527"]. Access the search results id column via list(search_results.id).

required
collection str

The data collection corresponding to the image ids.

required
output_directory Union[str, Path, None]

The file output directory, defaults to the current working directory.

None

Returns:

Type Description
List[str]

List of quicklook image output file paths.

Source code in up42/catalog.py
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
def download_quicklooks(
    self,
    image_ids: List[str],
    collection: str,
    output_directory: Union[str, Path, None] = None,
) -> List[str]:
    """
    Gets the quicklooks of scenes from a single sensor. After download, can
    be plotted via catalog.map_quicklooks() or catalog.plot_quicklooks().

    Args:
        image_ids: List of provider image_ids e.g. ["6dffb8be-c2ab-46e3-9c1c-6958a54e4527"].
            Access the search results id column via `list(search_results.id)`.
        collection: The data collection corresponding to the image ids.
        output_directory: The file output directory, defaults to the current working
            directory.

    Returns:
        List of quicklook image output file paths.
    """
    if self.data_products is None:
        self.data_products = self.get_data_products(basic=True)  # type: ignore
    host = [
        v["host"]
        for v in self.data_products.values()  # type: ignore
        if v["collection"] == collection
    ]
    if not host:
        raise ValueError(
            f"Selected collections {collection} is not valid. See catalog.get_collections."
        )
    host = host[0]
    logger.info(f"Downloading quicklooks from provider {host}.")

    if output_directory is None:
        output_directory = Path.cwd() / f"project_{self.auth.project_id}/catalog"
    else:
        output_directory = Path(output_directory)
    output_directory.mkdir(parents=True, exist_ok=True)
    logger.info(f"Download directory: {str(output_directory)}")

    if isinstance(image_ids, str):
        image_ids = [image_ids]

    out_paths: List[str] = []
    for image_id in tqdm(image_ids):
        try:
            url = (
                f"{self.auth._endpoint()}/catalog/{host}/image/{image_id}/quicklook"
            )
            response = self.auth._request(
                request_type="GET", url=url, return_text=False
            )
            out_path = output_directory / f"quicklook_{image_id}.jpg"
            out_paths.append(str(out_path))
            with open(out_path, "wb") as dst:
                for chunk in response:
                    dst.write(chunk)
        except ValueError:
            logger.warning(
                f"Image with id {image_id} does not have quicklook available. Skipping ..."
            )

    self.quicklooks = out_paths  # pylint: disable=attribute-defined-outside-init
    return out_paths

estimate_order(order_parameters, **kwargs)

Estimate the cost of an order.

Parameters:

Name Type Description Default
order_parameters Union[dict, None]

A dictionary like {dataProduct: ..., "params": {"id": ..., "aoi": ...}}

required

Returns:

Name Type Description
int int

An estimated cost for the order in UP42 credits.

Warning "Deprecated order parameters" The use of the 'scene' and 'geometry' parameters for the data estimation is deprecated. Please use the new order_parameters parameter as described above.

Source code in up42/catalog.py
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def estimate_order(self, order_parameters: Union[dict, None], **kwargs) -> int:
    """
    Estimate the cost of an order.

    Args:
        order_parameters: A dictionary like {dataProduct: ..., "params": {"id": ..., "aoi": ...}}

    Returns:
        int: An estimated cost for the order in UP42 credits.

    Warning "Deprecated order parameters"
        The use of the 'scene' and 'geometry' parameters for the data estimation is deprecated. Please use the new
        order_parameters parameter as described above.
    """
    if "scene" in kwargs or "geometry" in kwargs:
        # Deprecated, to be removed, use order_parameters.
        message = (
            "The use of the 'scene' and 'geometry' parameters for the data estimation is deprecated. "
            "Please use the new 'order_parameters' parameter."
        )
        warnings.warn(message, DeprecationWarning, stacklevel=2)
    elif order_parameters is None:
        raise ValueError("Please provide the 'order_parameters' parameter!")
    return Order.estimate(self.auth, order_parameters)  # type: ignore

search(search_parameters, as_dataframe=True)

Searches the catalog for the the search parameters and returns the metadata of the matching scenes.

Parameters:

Name Type Description Default
search_parameters dict

The catalog search parameters, see example.

required
as_dataframe bool

return type, GeoDataFrame if True (default), FeatureCollection if False.

True

Returns:

Type Description
Union[GeoDataFrame, dict]

The search results as a GeoDataFrame, optionally as JSON dict.

Example
    search_parameters={
        "datetime": "2019-01-01T00:00:00Z/2019-01-15T23:59:59Z",
        "collections": ["phr"],
        "intersects": {
            "type": "Polygon",
            "coordinates": [[[13.32113746,52.73971768],[13.15981158,52.2092959],
            [13.62204483,52.15632025],[13.78859517,52.68655119],[13.32113746,
            52.73971768]]]},
        "limit": 10,
        "sortby": [{"field" : "properties.acquisitionDate", "direction" : "asc"}]
        }
Source code in up42/catalog.py
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
def search(
    self, search_parameters: dict, as_dataframe: bool = True
) -> Union[GeoDataFrame, dict]:
    """
    Searches the catalog for the the search parameters and returns the metadata of
    the matching scenes.

    Args:
        search_parameters: The catalog search parameters, see example.
        as_dataframe: return type, GeoDataFrame if True (default), FeatureCollection if False.

    Returns:
        The search results as a GeoDataFrame, optionally as JSON dict.

    Example:
        ```python
            search_parameters={
                "datetime": "2019-01-01T00:00:00Z/2019-01-15T23:59:59Z",
                "collections": ["phr"],
                "intersects": {
                    "type": "Polygon",
                    "coordinates": [[[13.32113746,52.73971768],[13.15981158,52.2092959],
                    [13.62204483,52.15632025],[13.78859517,52.68655119],[13.32113746,
                    52.73971768]]]},
                "limit": 10,
                "sortby": [{"field" : "properties.acquisitionDate", "direction" : "asc"}]
                }
        ```
    """
    logger.info(f"Searching catalog with search_parameters: {search_parameters}")

    # The API request would fail with a limit above 500, thus 500 is forced in the initial
    # request but additional results are handled below via pagination.
    try:
        max_limit = search_parameters["limit"]
    except KeyError:
        logger.info("No `limit` parameter in search_parameters, using default 500.")
        max_limit = 500

    if max_limit > 500:
        search_parameters["limit"] = 500

    # UP42 API can query multiple collections of the same host at once.
    if self.data_products is None:
        self.data_products = self.get_data_products(basic=True)  # type: ignore
    hosts = [
        v["host"]
        for v in self.data_products.values()  # type: ignore
        if v["collection"] in search_parameters["collections"]
    ]
    if not hosts:
        raise ValueError(
            f"Selected collections {search_parameters['collections']} are not valid. See "
            f"catalog.get_collections."
        )
    if len(set(hosts)) > 1:
        raise ValueError(
            "Only collections with the same host can be searched at the same time. Please adjust the "
            "collections in the search_parameters!"
        )
    host = hosts[0]

    url = f"{self.auth._endpoint()}/catalog/hosts/{host}/stac/search"
    response_json: dict = self.auth._request("POST", url, search_parameters)
    features = response_json["features"]

    # Search results with more than 500 items are given as 50-per-page additional pages.
    while len(features) < max_limit:
        pagination_exhausted = len(response_json["links"]) == 1
        if pagination_exhausted:
            break
        next_page_url = response_json["links"][1]["href"]
        response_json = self.auth._request("POST", next_page_url, search_parameters)
        features += response_json["features"]

    features = features[:max_limit]
    if not features:
        df = GeoDataFrame(columns=["geometry"], geometry="geometry")
    else:
        df = GeoDataFrame.from_features(
            FeatureCollection(features=features), crs="EPSG:4326"
        )

    logger.info(f"{df.shape[0]} results returned.")
    if as_dataframe:
        return df
    else:
        return df.__geo_interface__