Tutorial: BrightEarth On Demand API
Introduction
BrigthEarth On Demand (BEOD) can be used using the Web application but can also be accessed by an API.
This API provides programmatic access to all functionalities of the BEOD platform.
In this tutorial you will learn how to, using the BEOD API, perform a geospatial data extraction over a user-defined area of interest (AOI).
Run your first extraction
In this section you will learn what are the different steps needed to create an extraction and how to programmatically achieve them using the API.
Authentication
The entrypoint of the API is https://api.beod.luxcarta.cloud.
To use the API, the client must be registred in the application.
We will assume in this tutorial that the client has already done the registration process via the Web application.
The BEOD API uses token-based authentication.
The API provides an endpoint allowing to get an authentication token (providing user email and password).
This token must be provided to all other endpoints of the API.
Getting a token and using it
To get a token the client must do a POST request to the /auth/login endpoint, providing in the body of the POST the user email and the user password.
Along other properties, the response contains the authentication token (access_token field).
Python example:
import sys
import requests
api_url = "https://api.beod.luxcarta.cloud"
def login(username: str, password: str) -> dict:
"""
This function logs in the user and returns the access token
Args:
username: str: The username of the user
password: str: The password of the user
Returns:
dict: A dict continaing the following entries:
{
"refresh_token": "eyJjdHkiOiJKV1QiLCJlbmMiOiJB...",
"access_token": "eyJraWQiOiJmK2czdXc4R05aZEc0WGs...",
"expires_in": 86400,
"token_type": "Bearer"
}
if the login was successful, None otherwise
"""
url = api_url + "/auth/login"
data = {
"email": username,
"password": password
}
response = requests.post(url, json=data)
if response.status_code == 200:
return response.json()
else:
return None
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
login_info = login(user, password)
if login_info:
token = login_info["access_token"]
print(token)
else:
print("Invalid credentials")
The /auth/login endpoint is the only one were the client have to provide the user password.
To call any other endpoints of the API, the client will have to include the returned access token in the header of the request.
The token is provided as a bearer in the Authorization property of the request header:
"Authorization": "Bearer the_token"
The following example illustrates how to use the token: a call is done to the /users/me endpoint which returns information on the user associated to the token. The programs then prints the id and the credits of the user.
Python example:
import sys
import requests
from authentication import login, api_url
def get_user(token):
url = api_url + "/users/me"
# use the token as a bearer token in the headers
headers = {
"Authorization": f"Bearer {token}"
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
return response.json()
else:
return None
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
token = login(user, password)["access_token"]
user_info = get_user(token)
if user_info:
print(f"User id: {user_info['id']}" )
print(f"Remaining credits: {user_info['credits']} )
else:
print("Invalid token")
Creation of a new project with an AOI
Once identified against the API, the next step is to create a new BEOD Project.
For this we will POST to the create project endpoint /users/{user_id}/projects.
Note: We can specify the real user id (which is a UUID, see example above) or use the generic user id me which always relate to the user associated with the used token.
So, let's create the project; the only property we have to provide is the name of a project. The endpoint will return the id of the newly created project.
Python example:
import sys
import requests
from pprint import pprint
from authentication import login, api_url
def create_project(token: str, project_name:str) -> int:
"""
Create a project with the given name.
Args:
token: str: The access token of the user
project_name: str: The name of the project
Returns:
int: The id of the created project if the project was created successfully, None otherwise
"""
url = api_url + "/users/me/projects"
# use the token as a bearer token in the headers
headers = {
"Authorization": f"Bearer {token}"
}
data = {
"name": project_name
}
response = requests.post(url, json=data, headers=headers)
# Note:
# - in case of success a 201 status code is returned (which means created)
# - the id of the created project is returned in the response
if response.status_code == 201:
return response.json()["project_id"]
else:
return None
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
token = login(user, password)["access_token"]
project_id = create_project(token, "Tutorial project")
if project_id:
print(f"The id of the newly created project is: {project_id}")
else:
print("Project creation failed")
Once the project is created we have to create an AOI asset to specify our area of interest. This the region we want to extract data over.
To create the AOI we have to POST to the /users/me/projects/{project_id}/assets/aoi endpoint.
The only mandatory parameter is the geometry of the AOI, it is specified as a WKT Polygon (with geographic coordinates, lon/lat ordering).
Python example:
import sys
import requests
from pprint import pprint
from authentication import login, api_url
from create_project import create_project
# here is the geometry of the AOI (a small area around our french office)
aoi_wkt_geometry = """
POLYGON ((
6.9526850935291 43.6127384923652,
6.94943181004093 43.6123795674845,
6.94656582220611 43.611370079773,
6.95060918882712 43.6056268939605,
6.95587640971274 43.605649329347,
6.95759600241363 43.6079825638504,
6.95877338119982 43.6099231178373,
6.95855649563394 43.6117963099847,
6.9526850935291 43.6127384923652
))"""
def create_aoi(token: str, project_id:str, wkt_geometry: str) -> int:
"""
Create an aoi with the given geometry for the specified project.
Args:
token: str: The access token of the user
project_id: int: The id of the project
wkt_geometry: str: The wkt geometry of the aoi
Returns:
int: The id of the created aoi if the aoi was created successfully, None otherwise
"""
url = api_url + f"/users/me/projects/{project_id}/assets/aoi"
# use the token as a bearer token in the headers
headers = {
"Authorization": f"Bearer {token}"
}
data = {
"geometry": wkt_geometry
}
response = requests.post(url, json=data, headers=headers)
# Note:
# - in case of success a 201 status code is returned (which means created)
# - the id of the created asset aoi is returned in the response
if response.status_code == 201:
return response.json()["asset_id"]
else:
return None
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
token = login(user, password)["access_token"]
project_id = create_project(token, "Tutorial project")
aoi_id = create_aoi(token, project_id, aoi_wkt_geometry)
if aoi_id:
print(f"The id of the newly created aoi is: {aoi_id}")
else:
print("AOI creation failed")
Running an Extraction Scenario
So far, we have created a project and an AOI for this project. Note that a project has only one associated AOI.
Any scenario that we will run in the frame of a project, will be run over this AOI.
We are now ready to run our first data extraction over this AOI.
To start the extraction scenario, we have to call the /users/me/projects/{project_id}/scenarios/{scenario_type} endpoint.
There several scenario types available; here we want to do a simple mono extraction over the ESRI basemap.
We have to provide at least the following parameters :
-
user_id (in url): we will use the generic user me,
-
project_id (in url): the project to which the scenario belongs,
-
scenario_type (in url): we will specify extraction-mono as we want to do a simple mono extraction,
-
name: this is the client-defined scenario name, any string will fit,
-
source_image_type: here we want to extract over the global ESRI basemap. We will specify basemap,
-
source_image_id: when the type is set to basemap, the id here is the id of the basemap. The id of the ESRI basemap is 1,
-
augmented: we will specify false as we don't want to augment the output buildings,
-
output_crs: this is the epsg code of the SRS to use for the output data, we will use 3857 which corresponds to the Global Mercator projection. The other possible value is 4326 which corresponds to WGS84 geographic coordinates,
-
vector_formats: this is the list of the vector formats to use for vector output layers; we can specify several. For example here we will request for ESRI shapefile and GeoPackage : shapefile,geopackage,
-
raster_formats: this is the list of the raster formats to use for raster output layers; we will specify GeoTIFF: geotiff,
-
three_d_formats: this is the 3D format tu use for buildings & trees. The only supported format for non augmented scenarios is 3d_tiles,
-
dl_models: we can specify alternate deep learning models, we will use default ones and specify [].
Python example:
import sys
import requests
from pprint import pprint
from authentication import login, api_url
from create_project import create_project
from create_aoi import create_aoi, aoi_wkt_geometry
def create_simple_mono_extraction_scenario(token: str, project_id:str) -> int:
"""
Create a simple mono extraction scenario in the given project.
Args:
token: str: The access token of the user
project_id: str: The id of the project
Returns:
tuple(int, int): The id of the created scenario and the id of the associated activity,
None if the scenario creation failed
"""
url = api_url + f"/users/me/projects/{project_id}/scenarios/extraction-mono"
# use the token as a bearer token in the headers
headers = {
"Authorization": f"Bearer {token}"
}
data = {
"name": "My First extraction",
"source_image_type": "basemap",
"source_image_id": 1,
"augmented": False,
"output_crs": 3857,
"vector_formats": "shapefile,geopackage",
"raster_formats": "geotiff",
"three_d_formats": "3d_tiles",
"dl_models": []
}
response = requests.post(url, json=data, headers=headers)
# Note:
# - in case of success a 201 status code is returned (which means created)
# - the id of the created scenario and activity are returned in the response
if response.status_code == 201:
json_response = response.json()
return json_response["scenario_id"], json_response["activity_id"]
else:
return None
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
token = login(user, password)["access_token"]
project_id = create_project(token, "Tutorial project")
aoi_id = create_aoi(token, project_id, aoi_wkt_geometry)
ids = create_simple_mono_extraction_scenario(token, project_id)
if ids:
scenario_id, activity_id = ids
print(f"The id of the newly created scenario is: {scenario_id}")
print(f"The id of the associated activity is: {activity_id}")
else:
print("Scenario creation failed")
In the POST response if everything went well, we will have the id of the created scenario and the id of an activity that can be used to follow the progress of the execution. How to use it will be explained in the next section.
Following an Extraction Scenario
When a scenario is successfully created, the POST response contains the id of an activity.
An activity is the generic mechanism the platform uses to allow to follow to progress the execution of a scenario.
The client application can fetch the status and the progress of an activity using this endpoint : /users/{user_id}/projects/{project_id}/activities/{activity_id}.
The response of the GET request will contain several fields.
{
'project_id': 789,
'user_id': 'c235a4c4-4041-7088-d792-6c985bb01d72',
'type': 'SCENARIO-EXECUTION',
'progress': 0.0,
'state': 'PENDING',
'creation_date': '2024-08-22T13:19:27.358250',
'start_date': None,
'end_date': None,
'details': '',
'misc': {},
'scenario_id': 543,
'activity_id': 2292,
'id': 2292
}
The two most used ones are:
- progress which indicates the percentage of work done,
- state which indicates if the scenario is CREATED, PENDING, RUNNING, SUCCESSFUL or FAILED.
Python example:
import sys
import requests
import time
from pprint import pprint
from authentication import login, api_url
from create_project import create_project
from create_aoi import create_aoi, aoi_wkt_geometry
from create_scenario import create_simple_mono_extraction_scenario
def follow_activity(token: str, project_id:str, activity_id: int) -> int:
"""
Follow an activity in a project until it terminates.
Args:
token: str: The access token of the user
project_id: str: The id of the project
activity_id: int: The id of the activity to follow
Returns:
True if the activity terminated successfully, False otherwise
"""
url = api_url + f"/users/me/projects/{project_id}/activities/{activity_id}"
# use the token as a bearer token in the headers
headers = {
"Authorization": f"Bearer {token}"
}
while True:
response = requests.get(url, headers=headers)
if response.status_code != 200:
print('Unable to track the activity')
return False
# get status and progress
json_response = response.json()
state = json_response["state"]
progress = json_response["progress"]
print(f'> Activity state: {state}, progress: {progress}')
if state == "SUCCESSFUL":
return True
elif state == "FAILED":
return False
time.sleep(5)
else:
return None
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
token = login(user, password)["access_token"]
project_id = create_project(token, "Tutorial project")
aoi_id = create_aoi(token, project_id, aoi_wkt_geometry)
scenario_id, activity_id = create_simple_mono_extraction_scenario(token, project_id)
if follow_activity(token, project_id, activity_id):
print("Scenario terminated successfully")
else:
print("Scenario terminated with an error")
Downloading results
If this scenario terminates properly we should be able to display the results from the web application.
In this section you will learn how to download the output files to be able to use them locally.
Depending on the size of the AOI, the results are zipped into a single file that can be downloaded at once or are not zipped and each file must be downloaded separately.
Regarding the size of the AOI used in this tutorial, a ZIP file is created, so the returned link will point to the ZIP. Even if the zip file is created it is still possible to download all the files separately. We will cover both cases in this tutorial.
The endpoint /users/{user_id}/projects/{project_id}/scenarios/{scenario_id}/download/urls will allow us to get the files to download, and for each one we will have a presigned url allowing us to download it securely from the backend storage.
If a zip file was generated, the response of the endpoint will look like:
{
"<scenario_id>.zip": "<url for downloading the zip>"
}
If no zip file was generated, we will get a list of files to download separately and the response of the endpoint is a nested dictionary replicating the backend storage folder structure. Each entry in the dict is either a file (the key is the file name, and the value the url), or a nested dict representing a folder.
Typically, the AOI is processed in tiles, so, the top-level folder contains a folder per tile, and, the top-level entries of the returned dictionnary are the tiles folders (tile names are numbers starting from 0):
{
"0": {
...
},
"1": {
...
},
...
}
Under each tile there is an EXPORT folder, containing a folder per layer (building, tree, road, clutter, dtm).
In each layer folder, we will find one sub-folder per selected format (shapefile, geopackage, geotiff, ...), and in this format folder we will find files.
{
"0": {
"EXPORT" : {
"building": {
"shapefile": {
"building.shp": "<the url to download building.shp>",
"building.shx": "<the url to download building.shx>",
"building.dbf": "<the url to download building.dbf>",
"building.prj": "<the url to download building.prf>",
},
"shapefile": {
"building.gpkg": "<the url to download building.gpkg>",
},
},
"tree" : {
"shapefile": {
...
}
},
...
},
"1" : {
"EXPORT" : {
"building": {
...
},
...
}
},
...
}
As there can be a large number of tiles, the endpoint may returns an incomplete list. In this case, the dict contains an addtional top-level key NextContinuationToken that can be used as a query parameter for a next request to the same endpoint. Each consecutive request, made with the continuation token returned by the previous request, will return an addtional chunk of files to download.
The query parameter to add to the request is continuation_token.
In this case the endpoint url looks like :
/users/{user_id}/projects/{project_id}/scenarios/{scenario_id}/download/urls&continuation_token=<the continuation token>
The following example will show how to call the download endpoint, how to interpret the returned results and how to download the files. It works for both cases ie. with and without zip. The local structure after the download will have the same structure than the backend storage.
Python example:
import sys
import shutil
from os import makedirs
from os.path import join, dirname
import requests
import time
from pprint import pprint
from authentication import login, api_url
def download_one_file(url: str, output_file: str):
"""
Download one file from the given url to the given output file.
Args:
url: str: The url of the file to download
output_file: str: The path of the output file
"""
print(f'Downloading {url} to {output_file}')
# ensure output folder exists
makedirs(dirname(output_file), exist_ok=True)
# stream download into the local file
with requests.get(url, stream=True) as response:
with open(output_file, 'wb') as f:
shutil.copyfileobj(response.raw, f)
def download_file_tree(files: dict, current_path = ""):
"""
Download the files in the given file tree.
Args:
files: dict: The file tree to download
current_path: str: The current path in the file tree
"""
for fileurl_or_folder in files.keys():
if isinstance(files[fileurl_or_folder], str):
# this a file url
download_one_file(files[fileurl_or_folder], join(current_path, fileurl_or_folder))
else:
# this is a sub-folder
download_file_tree(files[fileurl_or_folder], join(current_path, fileurl_or_folder))
def download_scenario(token: str, project_id:str, scenario_id: int, output_folder: str) -> int:
"""
Download the files of the given scenario to the given output folder.
Args:
token: str: The access token of the user
project_id: int: The id of the project
scenario_id: int: The id of the scenario
output_folder: str: The path of the output folder
Returns:
bool: True if the download was successful, False otherwise
"""
url = api_url + f"/users/me/projects/{project_id}/scenarios/{scenario_id}/download/urls"
continuation_token = ''
current_url = url
while True:
response = requests.get(url, headers={"Authorization": f"Bearer {token}"})
if response.status_code != 200:
return False
json_response = response.json()
download_file_tree(json_response, output_folder)
if "NextContinuationToken" in json_response:
current_url = url + f'&continuation_token={json_response["NextContinuationToken"]}'
else:
break
return True
if __name__ == "__main__":
user = sys.argv[1]
password = sys.argv[2]
project_id = sys.argv[3]
scenario_id = sys.argv[4]
output_folder = sys.argv[5]
user_info = login(user, password)
token = user_info["access_token"]
if download_scenario(token, project_id, scenario_id, output_folder):
print("Download successful")
else:
print("Download failed")
Conclusion
In this tutorial you learned how to:
-
identify yourself to the API and how to use the access token for the following requests,
-
create a new project and associate an AOI with it,
-
run and follow an extraction scenario,
-
download the results of the scenario.
Going further
The BEOD API contains many more endpoints allowing for example to retrieve the projects associated to a user, to retrieve scenarios and assets associated to a project. All these endpoint are described in the reference documentation of the API.
More tutorials concerning the API will be added soon.