Home / Python SDK / Sample notebooks

Track outputs related to a catalog order

Assign a unique identifier to an order to link the specific output back to its original order.

Open notebook

Assign a unique identifier to an order to link the specific output back to its original order. You’ll be able to do the following:

  • Track the order’s status.
  • Retrieve the delivered data.
  • Use the data as input for a processing job.
  • Find all data related to the processing job (both input and output).
Python
# Run this cell to set your unique order identifier
UNIQUE_IDENTIFIER_NAME = "Example_Unique_Identifier_123456"

Step 1. Set up the notebook

1. Install dependencies

Python
!pip install up42-py --upgrade -q


import up42, geojson, pathlib
from up42 import processing_templates
from itertools import islice
from datetime import date, timedelta
from IPython.display import Image

2. Configure credentials

Run the cell below to create a credentials.json file in a directory named .up42 in your home folder.

Python
# Defines the credentials file path if it doesn't exist
credentials_file_path = pathlib.Path.home().joinpath(".up42/credentials.json")
credentials_file_path.parent.mkdir(parents=True, exist_ok=True)
credentials_file_path.touch(exist_ok=True)


# Prints the path to the file
print(f"Credentials file is located at: {credentials_file_path}")
  1. Click the link above to the created file and paste the following code:
    JSON
    {
        "username": "<your-email-address>",
        "password": "<your-password>"
    }
  2. Retrieve the email address and password used for logging into the console. Use them as values for username and password.
  3. Save the credentials.json file.

3. Authenticate and create a PySTAC client connection

Python
up42.authenticate(cfg_file=credentials_file_path)


# Use region if you're logging with the NSG service
# up42.authenticate(cfg_file=credentials_file_path, region="sa")


catalog = up42.catalog_client()

Step 2. Find and select data

1. Search the catalog for scenes

Define the type of catalog scene you want to search for.

Python
# Select the collection
collection_name = "sentinel-2"


# Specify the maximum cloud cover percentage from 0 to 100
max_cloud_coverage_percentage = 10


# Define the time interval to search the last 6 months
end_date = date.today()
start_date = end_date - timedelta(days=180)


# Define search geometry
geometry = {
    "type": "Polygon",
    "coordinates": [
        [
            [13.369713, 52.452327],
            [13.369713, 52.470760],
            [13.339159, 52.470760],
            [13.339159, 52.452327],
            [13.369713, 52.452327],
        ]
    ],
}


# Search for scenes
items = list(
    catalog.search(
        collections=[collection_name],
        intersects=geometry,
        datetime=f"{start_date}/{end_date}",
        filter={
            "op": "<",
            "args": [{"property": "eo:cloud_cover"}, max_cloud_coverage_percentage],
        },
        max_items=100,
    ).items()
)

Output the scenes that match the specified parameters.

Python
print(f"\nFound {len(items)} scenes matching the criteria:\n")
for item in islice(items, 3):  # Print first 3
    print(f"- Scene ID:         {item.id}")
    print(f"  Acquisition date: {item.datetime}")
    print(f"  Cloud coverage:   {item.properties.get('eo:cloud_cover')}%")
    print(f"  Resolution:       {item.properties.get('gsd')} m\n")

2. Preview data

Some data hosts provide free, low-resolution quicklooks that you can view before creating an order:

  • Quicklooks are previews of full scenes. Some quicklooks may not fully match the scene they represent.
  • Thumbnails are small, light-weighted previews of full scenes. For some data hosts, thumbnails can be cropped and rotated compared to quicklooks.
Python
# Create an iterator to view quicklooks from search results
ql_items = (item for item in items if "quicklook" in item.assets)

Run this cell to view quicklooks from search results one by one:

Python
# Select the next scene that provides a quicklook
item = next(ql_items, None)
if item is None:
    raise RuntimeError("No scenes with a quicklook were found.")
print(f"Collection title:   {item.get_collection().title}")
print(f"Scene ID:           {item.id}")  # This is needed for ordering


# Download quicklook to quicklooks directory
quicklooks_dir = pathlib.Path("quicklooks")
quicklooks_dir.mkdir(exist_ok=True)
quicklook_path = item.assets["quicklook"].file.download(quicklooks_dir)
Image(quicklook_path)

3. Select a data product

Data products are a type of imagery with a specific geometric and radiometric processing level and file format. All collections have different data products corresponding to their technical specifications.

Python
# Provide scene ID or default to current quicklook
selected_id = input(f"Enter scene ID [{item.id}]: ").strip() or item.id
matches = [scene for scene in items if scene.id == selected_id]
if len(matches) != 1:
    raise RuntimeError(
        f"Expected one search result with scene ID {selected_id}, found {len(matches)}."
    )
item = matches[0]
collection = item.get_collection()
Python
# Find data products available for the scene previewed above
data_products = {
    name: asset.extra_fields["id"]
    for name, asset in collection.assets.items()
    if "data" in (asset.roles or [])
}


for name in data_products:
    asset = collection.assets[name]
    print(f"-  Title:             {asset.title}")
    print(f"   Data product name: {name}")
    print(f"   Data product ID:   {asset.extra_fields.get('id')}")
    print(f"   Description:       {asset.description}\n")
Python
# Select a data product for ordering
data_product_name = "sentinel-2-level-2a"
if data_product_name not in data_products:
    available = ", ".join(data_products) or "none"
    raise KeyError(
        f"{data_product_name} is not a data product of this scene. Available products: {available}."
    )
data_product_id = data_products[data_product_name]

Step 3. Get access

Once you’ve found a scene you want to order, check the requirements of its collection and review its EULA.

1. Request access

Some collections require an access approval before you can order them. Check if the collection of the chosen scene requires one.

Python
# Retrieve the selected collection
print(f"Collection ID: {collection.id}")
print(f"Collection title: {collection.title}")


if collection.extra_fields.get("access:approval_required"):
    print("You need to request access first")
else:
    print("No access request is required")

If needed, request access to the collection. Your request will be reviewed in up to 3 working days. You’ll get an email once a decision has been reached.

2. Accept the EULA

If you want to order the collection for the first time, review and accept its end-user license agreement (EULA). If a EULA is updated, you need to re-accept it before creating your next order.

Step 4. Create an order

1. Fill out an order form

Create an order template for the chosen scene and data product. The unique identifier is the order display name.

Python
features = geojson.FeatureCollection(features=[geojson.Feature(geometry=geometry)])
order_template = up42.BatchOrderTemplate(
    data_product_id=data_product_id,
    display_name=UNIQUE_IDENTIFIER_NAME,
    features=features,
    params={"id": item.id},
    tags=["sdk", data_product_name],
)

2. Estimate the cost

Before creating an order, estimate the price in UP42 credits. The factors considered in the final price calculation are the size of the area of interest, the data collection, and the type of data product.

Python
order_template.estimate

The response returns the overall credit amount that will be deducted from your credit balance if you decide to proceed with the ordering.

3. Create an order

Create an order with the requested parameters.

Credits will be deducted upon successful completion of the created catalog order. The transaction can’t be reversed.

Python
order_references = order_template.place()
order_reference = order_references[0]
if not hasattr(order_reference, "order"):
    raise RuntimeError(order_reference.message)
order = order_reference.order
order

Step 5. Monitor and download

To learn about the time frames of catalog orders, see Asset delivery time.

Check the status of your order. The following order statuses mean you can download assets from storage:

  • BEING_FULFILLED Some order assets might have been delivered.
  • FULFILLED All order assets have been delivered.
Python
order.status

You can also track the order until it is fulfilled. Tracking waits until the status is FULFILLED and raises an error if the order fails or is canceled.

Python
order.track(report_time=150)

Once an order is fulfilled, the data is delivered to your UP42 storage. Search storage by the order id. Use the delivered asset that has the data role, and set its up42-user:title to the unique identifier.

Python
# Initialize the STAC client to search your UP42 storage
UP42_client = up42.stac_client()


# Find assets delivered for this order
order_filter = {
    "op": "=",
    "args": [
        {"property": "up42-order:id"},
        order.id,
    ],
}


delivered_items = list(UP42_client.search(filter=order_filter).items())
if not delivered_items:
    raise RuntimeError(f"No delivered assets found for order {order.id}.")


data_items = [
    delivered
    for delivered in delivered_items
    if any("data" in (asset.roles or []) for asset in delivered.assets.values())
]
if not data_items:
    found = ", ".join(delivered.id for delivered in delivered_items)
    raise RuntimeError(
        f"No delivered asset with a data role for order {order.id}. Delivered items: {found}."
    )
if len(data_items) > 1:
    print(f"Found {len(data_items)} data assets. The first one is used for processing.")
stac_item = data_items[0]


# Keep the unique identifier on the delivered scene
stac_item.up42.title = UNIQUE_IDENTIFIER_NAME
stac_item.update()
print(f"Found delivered item with ID: {stac_item.id}")

Step 6. Process the data

1. Validate and estimate a processing job

Once you have your input data, you can run a processing job on it. Validate and estimate your job.

Python
# Create a unique title for the processing job's output
job_title = f"{UNIQUE_IDENTIFIER_NAME}_processing_output"


# Create the job template using our STAC item as input
job_template = processing_templates.UpsamplingNSSentinel(
    title=job_title, item=stac_item
)


# Stop when the job template is invalid
if not job_template.is_valid:
    for error in job_template.errors:
        print(error)
    raise RuntimeError("Processing job template is invalid.")


print(f"The estimated cost is {job_template.cost.credits} credits.")

2. Run a processing job

Python
# Execute the job and track its progress
job = job_template.execute()
job.track(wait=20)


print("\nTracking complete!")
print(f"The final job status is: {job.status.value}")


if job.status.value not in ("successful", "captured"):
    raise RuntimeError(f"Processing job finished with status {job.status.value}.")

3. Retrieve the processing output

Read the STAC items from the finished job’s output collection.

Python
if job.collection is None:
    raise RuntimeError("The processing job has no output collection.")


processed_items = list(UP42_client.search(collections=[job.collection.id]).items())
if not processed_items:
    raise RuntimeError("The processing job did not return any STAC items.")


processed_output_stac_item = processed_items[0]
print(f"Found processed item with ID: {processed_output_stac_item.id}")

Update the metadata of the processed STAC item, adding our original UNIQUE_IDENTIFIER_NAME to its up42-user:title field. This ensures that both the input and output assets share a common identifier.

Python
# Set the "up42-user:title" to link this output back to the original order
for processed_output_stac_item in processed_items:
    processed_output_stac_item.up42.title = UNIQUE_IDENTIFIER_NAME
    processed_output_stac_item.update()
    print(f"STAC item ID:   {processed_output_stac_item.id}")
    print(f"User title:     {processed_output_stac_item.up42.title}\n")

Perform a single search to retrieve both the original image and its processed output. The following search looks for any STAC item whose up42-user:title is the unique identifier.

Python
title_filter = {
    "op": "=",
    "args": [
        {"property": "up42-user:title"},
        UNIQUE_IDENTIFIER_NAME,
    ],
}


stac_items_search = UP42_client.search(filter=title_filter)


# This list includes both the processed output and the delivered scene
stac_items_list = list(stac_items_search.items())


print(
    f"Found {len(stac_items_list)} related items using the unique identifier '{UNIQUE_IDENTIFIER_NAME}':\n"
)
for item in stac_items_list:
    print(f"-   Item ID: {item.id}")
    print(f"    Filename: {item.properties['title']}")
    print(f"    User title: {item.properties['up42-user:title']}\n")