Skip to content

Tasking

The Tasking class enables access to the UP42 tasking functionality.

Use tasking:

tasking = up42.initialize_tasking()

Source code in up42/tasking.py
class Tasking(CatalogBase):
    """
    The Tasking class enables access to the UP42 tasking functionality.

    Use tasking:
    ```python
    tasking = up42.initialize_tasking()
    ```
    """

    def __init__(self, auth: Auth):
        self.auth = auth
        self.type = "TASKING"

    def construct_order_parameters(
        self,
        data_product_id: str,
        name: str,
        acquisition_start: str,
        acquisition_end: str,
        geometry: Union[dict, Feature, FeatureCollection, list, GeoDataFrame, Polygon],
    ):
        """
        Helps constructing the parameters dictionary required for the tasking order. Each sensor has additional
        parameters that are added to the output dictionary with value None. The potential values for to select from
        are given in the logs, for more detail on the parameter use `tasking.get_data_product_schema()`.

        Args:
            data_product_id: Id of the desired UP42 data product, see `tasking.get_data_products`
            name: Name of the tasking order project.
            acquisition_start: Start date of the acquisition period, e.g. "2022-11-01"
            acquisition_end: End date of the acquisition period, e.g. "2022-11-01"
            geometry: Geometry of the area to be captured, default a Polygon. Allows Point feature for specific
                data products.

        Returns:
            The constructed order parameters dictionary.

        Example:
            ```python
            order_parameters = tasking.construct_order_parameters(
                data_product_id='647780db-5a06-4b61-b525-577a8b68bb54',
                name="My tasking order",
                acquisition_start=2022-11-01,
                acquisition_end=2022-12-01,
                geometry={'type': 'Polygon',
                   'coordinates': (((13.375966, 52.515068),
                     (13.375966, 52.516639),
                     (13.378314, 52.516639),
                     (13.378314, 52.515068),
                     (13.375966, 52.515068)),)}
                )
            ```
        """
        start_date, end_date = format_time_period(
            start_date=acquisition_start, end_date=acquisition_end
        ).split("/")

        order_parameters = {
            "dataProduct": data_product_id,
            "params": {
                "displayName": name,
                "acquisitionStart": start_date,
                "acquisitionEnd": end_date,
            },
        }

        schema = self.get_data_product_schema(data_product_id)
        logger.info(
            "See `tasking.get_data_product_schema(data_product_id)` for more detail on the parameter options."
        )
        order_parameters = autocomplete_order_parameters(order_parameters, schema)

        geometry = any_vector_to_fc(vector=geometry)
        if geometry["features"][0]["geometry"]["type"] == "Point":
            # Tasking (e.g. Blacksky) can require Point geometry.
            order_parameters["params"]["geometry"] = geometry["features"][0]["geometry"]  # type: ignore
        else:
            geometry = fc_to_query_geometry(
                fc=geometry, geometry_operation="intersects"
            )
            order_parameters["params"]["geometry"] = geometry  # type: ignore

        return order_parameters

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

Methods

construct_order_parameters(self, data_product_id, name, acquisition_start, acquisition_end, geometry)

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

Parameters:

Name Type Description Default
data_product_id str

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

required
name str

Name of the tasking order project.

required
acquisition_start str

Start date of the acquisition period, e.g. "2022-11-01"

required
acquisition_end str

End date of the acquisition period, e.g. "2022-11-01"

required
geometry Union[dict, geojson.feature.Feature, geojson.feature.FeatureCollection, list, geopandas.geodataframe.GeoDataFrame, shapely.geometry.polygon.Polygon]

Geometry of the area to be captured, default a Polygon. Allows Point feature for specific data products.

required

Returns:

Type Description

The constructed order parameters dictionary.

Examples:

order_parameters = tasking.construct_order_parameters(
    data_product_id='647780db-5a06-4b61-b525-577a8b68bb54',
    name="My tasking order",
    acquisition_start=2022-11-01,
    acquisition_end=2022-12-01,
    geometry={'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/tasking.py
def construct_order_parameters(
    self,
    data_product_id: str,
    name: str,
    acquisition_start: str,
    acquisition_end: str,
    geometry: Union[dict, Feature, FeatureCollection, list, GeoDataFrame, Polygon],
):
    """
    Helps constructing the parameters dictionary required for the tasking order. Each sensor has additional
    parameters that are added to the output dictionary with value None. The potential values for to select from
    are given in the logs, for more detail on the parameter use `tasking.get_data_product_schema()`.

    Args:
        data_product_id: Id of the desired UP42 data product, see `tasking.get_data_products`
        name: Name of the tasking order project.
        acquisition_start: Start date of the acquisition period, e.g. "2022-11-01"
        acquisition_end: End date of the acquisition period, e.g. "2022-11-01"
        geometry: Geometry of the area to be captured, default a Polygon. Allows Point feature for specific
            data products.

    Returns:
        The constructed order parameters dictionary.

    Example:
        ```python
        order_parameters = tasking.construct_order_parameters(
            data_product_id='647780db-5a06-4b61-b525-577a8b68bb54',
            name="My tasking order",
            acquisition_start=2022-11-01,
            acquisition_end=2022-12-01,
            geometry={'type': 'Polygon',
               'coordinates': (((13.375966, 52.515068),
                 (13.375966, 52.516639),
                 (13.378314, 52.516639),
                 (13.378314, 52.515068),
                 (13.375966, 52.515068)),)}
            )
        ```
    """
    start_date, end_date = format_time_period(
        start_date=acquisition_start, end_date=acquisition_end
    ).split("/")

    order_parameters = {
        "dataProduct": data_product_id,
        "params": {
            "displayName": name,
            "acquisitionStart": start_date,
            "acquisitionEnd": end_date,
        },
    }

    schema = self.get_data_product_schema(data_product_id)
    logger.info(
        "See `tasking.get_data_product_schema(data_product_id)` for more detail on the parameter options."
    )
    order_parameters = autocomplete_order_parameters(order_parameters, schema)

    geometry = any_vector_to_fc(vector=geometry)
    if geometry["features"][0]["geometry"]["type"] == "Point":
        # Tasking (e.g. Blacksky) can require Point geometry.
        order_parameters["params"]["geometry"] = geometry["features"][0]["geometry"]  # type: ignore
    else:
        geometry = fc_to_query_geometry(
            fc=geometry, geometry_operation="intersects"
        )
        order_parameters["params"]["geometry"] = geometry  # type: ignore

    return order_parameters

get_collections(self) inherited

Get the available data collections.

Source code in up42/tasking.py
def get_collections(self) -> Union[Dict, List]:
    """
    Get the available data collections.
    """
    url = f"{self.auth._endpoint()}/collections"
    json_response = self.auth._request("GET", url)
    collections = [c for c in json_response["data"] if c["type"] == self.type]
    return collections

get_data_product_schema(self, data_product_id) inherited

Gets the schema of a data product to help with the construction of the catalog/tasking order parameters.

Parameters:

Name Type Description Default
data_product_id str

The id of a catalog/tasking data product.

required
Source code in up42/tasking.py
def get_data_product_schema(self, data_product_id: str):
    """
    Gets the schema of a data product to help with the construction of the catalog/tasking order parameters.

    Args:
        data_product_id: The id of a catalog/tasking data product.
    """
    url = f"{self.auth._endpoint()}/orders/schema/{data_product_id}"
    json_response = self.auth._request("GET", url)
    return json_response  # Does not contain usual "data" key

get_data_products(self, basic=True) inherited

Get the available data products (combination of collection and data configuration, e.g. Pleiades Display product) for catalog or tasking.

Parameters:

Name Type Description Default
basic bool

A dictionary containing only the collection title, name, host and available data product configurations, default True.

True
Source code in up42/tasking.py
def get_data_products(self, basic: bool = True) -> Union[Dict, List]:
    """
    Get the available data products (combination of collection and data configuration, e.g.
    Pleiades Display product) for catalog or tasking.

    Args:
        basic: A dictionary containing only the collection title, name, host and available
            data product configurations, default True.
    """
    url = f"{self.auth._endpoint()}/data-products"
    json_response = self.auth._request("GET", url)
    unfiltered_products: list = json_response["data"]

    products = []
    for product in unfiltered_products:
        if product["collection"]["type"] != self.type:
            continue
        try:
            if not product["collection"]["isIntegrated"]:
                continue
        except KeyError:  # isIntegrated potentially removed from future public API
            pass
        try:
            if not product["productConfiguration"]["isIntegrated"]:
                continue
        except KeyError:
            pass
        products.append(product)

    if not basic:
        return products
    else:
        collection_overview = {}
        for product in products:
            collection_title = product["collection"]["title"]
            collection_name = product["collectionName"]
            host = product["collection"]["host"]["name"]
            data_product = {product["productConfiguration"]["title"]: product["id"]}

            if collection_title not in collection_overview:
                collection_overview[collection_title] = {
                    "collection": collection_name,
                    "host": host,
                    "data_products": data_product,
                }
            else:
                # Add additional products for same collection
                collection_overview[collection_title]["data_products"][
                    product["productConfiguration"]["title"]
                ] = product["id"]

        return collection_overview

place_order(self, order_parameters, track_status=False, report_time=120, **kwargs) inherited

Place an order.

Parameters:

Name Type Description Default
order_parameters Optional[dict]

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

required
track_status bool

If set to True, will only return the Order once it is FULFILLED or FAILED.

False
report_time int

The interval (in seconds) to query the order status if track_status is True.

120

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

Warning: When placing orders of items that are in archive or cold storage, the order fulfillment can happen up to 24h after order placement. In such cases, please make sure to set an appropriate report_time. You can also use Order.track_status on the returned object to track the status later.

Returns:

Type Description
Order

The placed order.

Source code in up42/tasking.py
def place_order(
    self,
    order_parameters: Union[dict, None],
    track_status: bool = False,
    report_time: int = 120,
    **kwargs,
) -> "Order":
    """
    Place an order.

    Args:
        order_parameters: A dictionary like {dataProduct: ..., "params": {"id": ..., "aoi": ...}}
        track_status (bool): If set to True, will only return the Order once it is `FULFILLED` or `FAILED`.
        report_time (int): The interval (in seconds) to query the order status if `track_status` is True.

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

     Warning:
        When placing orders of items that are in archive or cold storage,
        the order fulfillment can happen up to **24h after order placement**.
        In such cases, please make sure to set an appropriate `report_time`.
        You can also use `Order.track_status` on the returned object to track the status later.

    Returns:
        Order: The placed order.
    """
    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 ordering 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!")

    order = Order.place(self.auth, order_parameters)  # type: ignore
    if track_status:
        order.track_status(report_time)
    return order