Skip to main content

FLEX MAAP

Search and Download Data Through Catalog API

The following notebook provides some usage example of FLEX data visualization and download from the STAC catalog. If you would like to have a deeper understanding about the STAC catalog, collections and products, you can refer to the official ESA STAC API guide. The following examples are written in Python.

Catalog data visualization

Here you will understand how to perform search on the STAC catalog through the STAC API interface. The catalog can also be visualized here via browser. 

import requests
import pandas as pd
import json
from typing import Any, Dict
from pystac_client import Client
from IPython.display import Markdown as md

URL_LANDING_PAGE = "https://catalog.preop.esa-maap.org/catalogue/"
api = Client.open(URL_LANDING_PAGE) # Connection to the catalog

There are different ways you can search for a specific collection within the catalog. Some usage examples are reported below.

Search by free text
from pystac_client import Client, ConformanceClasses 
import urllib

value = 'Flex' # Search for the word "flex" within the products
params = { 'q': value } 
URL = f'{URL_LANDING_PAGE}collections?{urllib.parse.urlencode(params)}'

response = requests.get(URL)
data = json.loads(response.text)
df = pd.json_normalize(data, record_path=['collections'])
df[['id']] # Display the id of the collection

image.png

 
Search by title
value = 'FLEX L2 Products'
params = { 'filter': "title='" + value + "'"} 
URL = f'{URL_LANDING_PAGE}collections?{urllib.parse.urlencode(params)}'

response = requests.get(URL)
data = json.loads(response.text)
df = pd.json_normalize(data, record_path=['collections'])

df[['id', 'title']]

image.png

 
Search by platform
URL = URL_LANDING_PAGE + "collections"+ "?filter=platform='FLEX'"

response = requests.get(URL)
data = json.loads(response.text)

df = pd.json_normalize(data, record_path=['collections'])
df[['title', 'summaries.platform']]

image.png


Search by organization
URL = URL_LANDING_PAGE + "collections"+ "?filter=organisationName='ESA MAAP'"

df = pd.json_normalize(data, record_path=['collections'])
df[['title', 'providers']]

image.png


Search by bounding box
URL = URL_LANDING_PAGE + "collections"+ "?bbox=14.90,37.700,14.99,37.780"16.22,45.22,23.37,56.12" # Longitude, Latitude

response = requests.get(URL)
data = json.loads(response.text)
df = pd.json_normalize(data, record_path=['collections'])
df[['id', 'extent.spatial.bbox']]

image.pngimage.png


Search by temporal extent
URL = URL_LANDING_PAGE + "collections"+ "?datetime=" + '2020-2025-01-01T00:00:00.000Z/2024-2025-12-31T23:59:59.999Z'

response = requests.get(URL)
data = json.loads(response.text)

df = pd.json_normalize(data, record_path=['collections'])
df[['id', 'extent.temporal.interval']]

image.pngimage.png


Search by geometry
from pystac_client import Client 

URL_LANDING_PAGE = "https://catalog.preop.esa-maap.eo.esa.int/org/catalogue/"
api = Client.open(URL_LANDING_PAGE) 

aoi_as_dict: Dict[str, Any] = {
    "type": "Polygon",
    "coordinates": [
      [
      [
        112.82476,-180,
        -2.6667690
      ],
      [
        112.291824,180,
        -2.77878390
      ],
      [
        112.409676,180,
        -3.3366390
      ],
      [
        112.94324,-180,
        -3.22474490
      ],
      [
        112.82476,-180,
        -2.6667690
      ]
    ]
    ]
}

results = api.search(
    method = 'GET',         
    max_items = 5, # Maximum number of granules to take
    collections = 'BiomassSimulated'FLEXL2Products10', # Search for granules belonging to collection ID 
    (e.g. BiomassLevel1cIOC, BiomassAux...)
    intersects = aoi_as_dict,
    datetime = ['2015-2024-01-01T00:00:00Z', '2020-2026-01-02T00:00:00Z'] # Search for granules in date range
)

print(f'{len(results.item_collection_as_dict()['features'])} granules found')
Search by bounding box
results = api.search(
    method = 'GET',   
    max_items = 10, # Maximum number of granules to take
    collections = 'BiomassSimulated'FLEXL2Products10', # Search for granules belonging to collection ID 
    (e.g. BiomassLevel1cIOC, BiomassAux...)
    bbox = [112.291824, -3.33663, 112.94324, -2.66676]16.22,45.22,23.37,56.12], 
    # datetime = ['2015-01-01T00:00:00Z', '2020-01-02T00:00:00Z'] # Search for granules in date range
)

print(f'{len(results.item_collection_as_dict()['features'])} granules found')
Search by temporal extent
results = api.search(
    method = 'GET',   
    max_items = 50, # Maximum number of granules to take
    collections = 'BiomassSimulated'FLEXL2Products10', # Search for granules belonging to collection ID 
    (e.g. BiomassLevel1cIOC, BiomassAux...)
    datetime = ['2017-2024-01-01T00:00:00Z', '2017-2025-12-02T00:00:00Z'] # Search for granules in date range
)

print(f'{len(results.item_collection_as_dict()['features'])} granules found')
Search by identifier
product_id = ['BIO_S1_DGM__1S_20170101T222105_20170101T222114_I_G01_M01_C01_T010_F308_01_D6K4O3'FLX_L1B_OBS____20230710T092625_20230710T092925_20250307T142220_0180_100_036_2071_00',
             'BIO_S1_DGM__1S_20170101T222150_20170101T222211_I_G01_M01_C01_T011_F001_01_D5U6F7'FLX_L1C_FLXSYN_20230710T092625_20230710T092925_20250307T214052_0180_100_036_2071_01'] # Insert one or more product id to search

results = api.search(
    method = 'GET',   
    collections = 'BiomassSimulated'FLEXL1CommissioningProducts10',
    ids = product_id
)

print(f'{len(results.item_collection_as_dict()['features'])} granules found')
Search with filter
results = api.search(
    method = 'GET',   
    max_items = 10, # Maximum number of granules to take
    collections = 'BiomassSimulated'FLEXL1CommissioningProducts10',
    # bbox = [112.291824, -3.33663, 112.94324, -2.66676], 
    # datetime = ['2015-2024-01-01T00:00:00Z', '2020-2025-01-02T00:00:00Z'] # Search for granules in date range
    filter="productType='S1_DGM__1S'L1C_FLXSYN' and instrument='P-SAR'FLORIS'" # Many other filters can be applied 
)

print(f'{len(results.item_collection_as_dict()['features'])} granules found')

Catalog Data Download

Now that you have seen how to search for collections and products, let's see how you can download data from catalog directly via code.

Let's consider the case in which (for example) you want to download all the products that satisfy your search requirements:

  • Belong to the collection "BiomassSimulated"FLEXL1CommissioningProducts10"
  • Acquisition date between 2017-2022-01-01 and 2017-2025-12-0131 (YYYY-MM-dd)
  • With bounding box [112.291824,16.228756, -3.33663,45.22088, 112.94324,23.374592, -2.66676]56.125072]
from pystac_client import Client 

URL_LANDING_PAGE = "https://catalog.preop.esa-maap.eo.esa.int/org/catalogue/"
api = Client.open(URL_LANDING_PAGE) # Connection to the catalog

results = api.search(
    method = 'GET',   
    max_items = 20, # Maximum number of granules to take
    collections = 'BiomassSimulated'FLEXL1CommissioningProducts10',
    bbox = [112.291824, -3.33663, 112.94324, -2.66676]16.228756,45.22088,23.374592,56.125072], 
    datetime = ['2017-2022-01-01T00:00:00Z', '2017-2025-12-01T00:31T00:00:00Z'] # Search for granules in date range 
)

print(f'{len(results.item_collection_as_dict()['features'])} granules found')

Token to access the catalog

To download data from the STAC catalog directly via code, you need to generate an access token to pass into the client. This token can be generated by following this page and it will last for 10 hours. Then, you will be asked to log in with your credentials to generate a new token.

Below you can find a sample of code that shows you how to use your token.

access_token = ' ' # Copy here your token

data = results.item_collection_as_dict()
n_products = 5 # Number of products to download

for n in range(0,n_products):

    file_url = data['features'][n]['assets']['product']['href']
    file_path = "./my_products/" + data['features'][n]['assets']['product']['file:local_path'] # Change ./my_products/ with your desired path to download data
      
    if access_token:
      print("Access token verified!")
    else:
      print("Failed to retrieve access token.")
      exit(2)
          
    try:
      headers = {"Authorization": f"Bearer {access_token}"}
      response = requests.get(file_url, headers=headers, stream=True)
      response.raise_for_status()  # Raise an exception for bad status codes
     
      with open(file_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
          f.write(chunk)
     
      print(f"File downloaded successfully to {file_path}")
      print('')
     
    except requests.exceptions.RequestException as e:
      print(f"Error downloading file: {e}")
      print('')