MAAP Python Library Function Guide (maap-py v5, OGC)
This notebook provides comprehensive documentation and examples for functions related to algorithms and jobs following OGC
[ ]:
from maap.maap import MAAP
import json
import os
import time
maap = MAAP()
Algorithm Management Functions
list_algorithms()
List all available OGC processes/algorithms.
[ ]:
# List all algorithms
response = maap.list_algorithms()
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
algorithms = response.json()
print(f"Found {len(algorithms.get('processes', []))} algorithms")
# Display first few algorithms
for i, algo in enumerate(algorithms.get('processes', [])[:3]):
print(f"{i+1}. ID: {algo.get('id')}, Title: {algo.get('title')}")
else:
print(f"Error: {response.text}")
Status Code: 200
Found 3 algorithms
1. ID: shah-sardem-sarsen, Title: shah-sardem-sarsen
2. ID: sardem-sarsen1, Title: sardem-sarsen
3. ID: mlucas-operawatermask1, Title: mlucas-operawatermask1
register_algorithm()
Deploy a new OGC process from a CWL file.
[ ]:
# Register algorithm from CWL file URL
# NOTE that you need to use the link with the raw version
cwl_url = "https://repo.dit.maap-project.org/root/ogc-application-packages/-/raw/main/mlucas/sardem-sarsen-mlucas/nasa-ogc/process.cwl?ref_type=heads"
response = maap.register_algorithm(execution_unit_href=cwl_url)
print(f"Registration Status: {response.status_code}")
if response.status_code == 202:
result_register_algorithm = response.json()
print(json.dumps(result_register_algorithm, indent=4))
else:
print(f"Error: {response.text}")
Registration Status: 202
{
"title": "sardem-sarsen-mlucas",
"description": "sardem-sarsen",
"keywords": [
"null"
],
"metadata": [],
"id": "sardem-sarsen-mlucas",
"version": "nasa-ogc",
"jobControlOptions": [],
"deploymentJobID": 4,
"status": "accepted",
"created": "2025-10-06 22:19:02.338271",
"links": [
{
"href": "/ogc/deploymentJobs/4",
"rel": "monitor",
"type": "application/json",
"hreflang": "en",
"title": "Deploying process status link"
}
],
"processPipelineLink": {
"href": "https://repo.dit.maap-project.org/root/deploy-ogc-hysds/-/pipelines/9901",
"rel": "monitor",
"type": "text/html",
"hreflang": "en",
"title": "Link to process pipeline"
}
}
get_deployment_status()
Check the status of an algorithm deployment.
[ ]:
# Check deployment status
deployment_id = result_register_algorithm.get("deploymentJobID")
print(f"Checking deployment status for deployment {deployment_id}")
response = maap.get_deployment_status(deployment_id)
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
deployment_info = response.json()
status = deployment_info["status"]
print(f"Deployment Status is: {status}")
else:
print(f"Error: {response.text}")
Checking deployment status for deployment 4
Status Code: 200
Deployment Status is: running
delete_algorithm()
Delete an algorithm (must be original deployer).
[ ]:
# Delete algorithm
# Change to be a process id for one of your processes
process_id = 5
response = maap.delete_algorithm(process_id)
print(f"Delete Status: {response.status_code}")
if response.status_code == 200:
print("Algorithm deleted successfully")
else:
print(f"Error: {response.text}")
Delete Status: 200
Algorithm deleted successfully
register_algorithm_from_cwl_file()
Register algorithm from a local CWL configuration file.
[ ]:
# This function has not been implemented yet
describe_algorithm()
Get detailed information about a specific algorithm.
[ ]:
# Describe a specific algorithm
process_id = 1
response = maap.describe_algorithm(process_id)
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
algorithm_info = response.json()
print(f"Algorithm Title: {algorithm_info.get('title')}")
print(f"Description: {algorithm_info.get('description')}")
print(f"Version: {algorithm_info.get('version')}")
else:
print(f"Error: {response.text}")
Status Code: 200
Algorithm Title: shah-sardem-sarsen
Description: This application is designed to process Synthetic Aperture Radar (SAR) data from Sentinel-1 GRD (Ground Range Detected) products using a Digital Elevation Model (DEM) obtained from Copernicus.
Version: nasa-ogc
update_algorithm()
Update an existing algorithm (must be original deployer).
[ ]:
# Update existing algorithm
process_id = 3
# Provide a new_cwl_url with the same id/version of the process you are trying to alter
new_cwl_url = "https://raw.githubusercontent.com/grallewellyn/cwl-files/refs/heads/main/cwl6.cwl"
response = maap.update_algorithm(process_id, new_cwl_url)
print(f"Update Status: {response.status_code}")
if response.status_code == 202:
result = response.json()
print(f"Algorithm updated successfully")
else:
print(f"Error: {response.text}")
Update Status: 202
Algorithm updated successfully
get_algorithm_package()
Get the formal package description for an algorithm.
[ ]:
# Get algorithm package
process_id = 1
response = maap.get_algorithm_package(process_id)
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
package_info = response.json()
print(f"Package description: {json.dumps(package_info, indent=2)}")
else:
print(f"Error: {response.text}")
Status Code: 200
Package description: {
"processDescription": "This application is designed to process Synthetic Aperture Radar (SAR) data from Sentinel-1 GRD (Ground Range Detected) products using a Digital Elevation Model (DEM) obtained from Copernicus.",
"executionUnit": {
"href": "https://repo.uat.maap-project.org/root/ogc-application-packages/-/raw/main/sshah/shah-sardem-sarsen/nasa-ogc/process.cwl",
"rel": "monitor-desc",
"type": "text/html",
"hreflang": "en",
"title": "Process Reference"
}
}
Job Management Functions
submit_job()
Execute an algorithm job.
[ ]:
# Submit a job
process_id = 1
inputs = {
"bbox": "-118.06817 34.22169 -118.05801 34.22822",
"stac_catalog_folder": "https://cmr.earthdata.nasa.gov/stac/ASF/collections/SENTINEL-1A_DP_GRD_HIGH_1/items/S1A_IW_GRDH_1SDV_20250330T171421_20250330T171446_058537_073E4F_985B-GRD_HD",
"stac_asset_name": "edu/GRD_HD/SA/S1A_IW_GRDH_1SDV_20250330T171421_20250330T171446_058537_073E4F_985B"
}
queue = "maap-dps-sandbox"
response = maap.submit_job(
process_id=process_id,
inputs=inputs,
queue=queue,
dedup=True, # Optional: prevent duplicate jobs
tag="my-experiment-run1" # Optional: user-defined tag
)
print(f"Job Submission Status: {response.status_code}")
if response.status_code == 202:
job_info = response.json()
print(f"Status: {job_info.get('status')}")
job_id = job_info.get('id')
print(f"Job ID: {job_info.get('id')}")
else:
print(f"Error: {response.text}")
get_job_status()
Check the status of a submitted job.
[ ]:
# Check job status
print(f"Checking job status for {job_id}")
response = maap.get_job_status(job_id)
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
job_status = response.json()
print(f"Job Status: {json.dumps(job_status, indent=2)}")
else:
print(f"Error: {response.text}")
get_job_result()
Get results from a completed job.
[ ]:
# Get job results
print(f"Checking job results for {success_job_id}")
response = maap.get_job_result(success_job_id)
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
results = response.json()
print(f"Job Results: {json.dumps(results, indent=2)}")
else:
print(f"Error: {response.text}")
cancel_job()
Cancel a running job or delete a queued job.
[ ]:
# Cancel job
process_id = existing_process_id_dont_alter
inputs = {
"bbox": "-118.06817 34.22169 -118.05801 34.22822",
"stac_catalog_folder": "https://cmr.earthdata.nasa.gov/stac/ASF/collections/SENTINEL-1A_DP_GRD_HIGH_1/items/S1A_IW_GRDH_1SDV_20250330T171421_20250330T171446_058537_073E4F_985B-GRD_HD",
"stac_asset_name": "edu/GRD_HD/SA/S1A_IW_GRDH_1SDV_20250330T171421_20250330T171446_058537_073E4F_985B"
}
queue = "maap-dps-sandbox"
response = maap.submit_job(
process_id=process_id,
inputs=inputs,
queue=queue,
dedup=False, # Optional: prevent duplicate jobs
tag="my-experiment-run1" # Optional: user-defined tag
)
job_info = response.json()
if response.status_code == 202:
job_id = job_info.get('id')
print(f"Waiting for job {job_id} to be accepted as its status so that it can be cancelled")
time.sleep(10)
print(f"Cancelling job with ID {job_id}")
# Cancel without waiting
response = maap.cancel_job(job_id)
print(f"Cancel Status: {response.status_code}")
print(response.text)
list_jobs()
List jobs with various filtering options.
[ ]:
# List all jobs (basic)
jobs_response = maap.list_jobs()
print(f"Status Code: {jobs_response.status_code}")
if jobs_response.status_code == 200:
jobs = jobs_response.json()
print(f"Found {len(jobs.get('jobs', []))} total user jobs")
# List jobs with filters
filtered_jobs_response = maap.list_jobs(
process_id=1,
status="failed",
limit=5,
page_size=10,
queue="maap-dps-sandbox",
tag="testing",
min_duration=60, # minimum 60 seconds
max_duration=3600, # maximum 1 hour
datetime="2023-01-01T00:00:00Z/2025-12-31T23:59:59Z"
)
print(f"Filtered jobs {json.dumps(filtered_jobs_response.json(), indent=2)}")
get_job_metrics()
Get performance metrics for a job.
[ ]:
# Get job metrics
response = maap.get_job_metrics(success_job_id)
print(f"Status Code: {response.status_code}")
if response.status_code == 200:
metrics = response.json()
print(f"Job Metrics: {json.dumps(metrics, indent=2)}")
else:
print(f"Error: {response.text}")
Summary
This notebook has covered the functions available in the MAAP Python library for algorithms and jobs:
Algorithm Management
list_algorithms()- List available algorithmsregister_algorithm()- Deploy new algorithmsdescribe_algorithm()- Get algorithm detailsupdate_algorithm()- Update existing algorithmsdelete_algorithm()- Delete algorithmsget_algorithm_package()- Get algorithm package infoget_deployment_status()- Check deployment status
Job Management
submit_job()- Execute algorithm jobsget_job_status()- Check job statusget_job_result()- Get job resultscancel_job()- Cancel running jobslist_jobs()- List user jobs with filtersget_job_metrics()- Get job performance metrics
Each function is designed to work with the MAAP platform’s OGC-compliant API