Usage

Design Philosophy

These APIs are intended to be used within code, with Publish and Subscription intended for automated live services, whilst Dataset and Item for manual use cases (for example, in a Jupyter Notebook). As such, Dataset has a complex endpoint to retrieve the data you want from your Dataset Packages. To make it easier to work with this endpoint, you can generate the endpoint via the Data Library in the Dashboard. This option is available for Subscription and Item as well, but their endpoints are straight forward in comparison.

You can use the APIs outside their intended use case, but these will not be covered here.

Note

Our examples primarily use Python, but these endpoints are language agnostic.

Publish

Basic

Here we use the websockets and orjson packages (these packages are preferences not requirements) to send STAC Items to the Publish endpoint. In the example, our Items are in a list called stac_items.

import websockets
import asyncio
import orjson

uri = "wss://ws.datasphere.astrodynamic.space/publish/v1/"
api_token = <API Token>

async def main():

    async with websockets.connect(uri=uri, extra_headers={"Authorization":f"Token {api_token}"}) as websocket:

        for stac_item in stac_items:
            message = orjson.dumps(
                {
                    "active_parent": "cool-sat",
                    "provider": "astro_dynamic",
                    "subscribable": false,
                    "data": <stac_item : dict>
                }
            )

        await websocket.send(message)

if __name__ == "__main__":
    asyncio.run(main)

Subscription

BasicGeneral

import requests
import json

uri = "wss://datasphere.astrodynamic.space/subscription/v1/?package={package name}"
api_token = <API Token>

with requests.Session().get(url, headers={"Authorization":f"Token {api_token}"}, stream=True) as response:
    for line in response.iter_lines():
        if line:
            stac_item = json.loads(line.decode())

Here we show how you could use the Subscription endpoint to get the assets required. This is a simple, general example.

import requests
import json
import pathlib

uri = "wss://datasphere.astrodynamic.space/subscription/v1/?package={package name}"
api_token = <API Token>
required_asset_keys = ["B02_10m","B03_10m","B04_10m"]
path = pathlib.Path().resolve()

def save_asset(response_content : str, item_id : str, asset_key : str):
    """
    Save the asset to the local directory.
    """
    file_path = path + f"/{item_id}-{asset_key}.tif"
    with open(file_path, 'wb') as f:
        f.write(response.content)
    return

def retrieve_asset(asset_href : str):
    """
    Retrieves asset from href.
    """
    retrieve_response = requests.get(asset_href)

    return retrieve_response.content

def extract_asset(stac_item : dict):
    """
    Extracts asset href from STAC Item Assets Object.
    """
    asset_objects : dict = stac_item.get("assets")
    item_id = stac_item.get("id")

    for asset_key, asset_object in asset_objects.items():
        if asset_key in required_asset_keys:
            asset_href = asset_object.get("href")
            asset_content = retrieve_asset()
            save_asset(asset_content, item_id, asset_key)
    return

def main():
    with requests.Session().get(url, headers={"Authorization":f"Token {api_token}"}, stream=True) as response:
        for line in response.iter_lines():
            if line:
                stac_item = json.loads(line.decode())
                extract_asset(stac_item)
    return

if __name__ == "__main__":
    main()

Dataset

StreamPaginate

Basic

    import requests
    import json

    session = requests.Session()
    stac_items = list()

    url = "https://datasphere.astrodynamic.space/api/dataset/v1/{package name}/stream/"

    with session.get(url, headers={"Authorization":"Token {api token}"}, stream=True) as response:
        for line in response.iter_lines():
            if line:
                stac_item = json.loads(line.decode())
                stac_items.append(stac_item)

Python

import requests

url = "https://datasphere.astrodynamic.space/api/dataset/v1/{package name}/paginate/"

response = request.get(url, headers={"Authorization":"Token {api token}"})
feature_collection = response.json()

Item

Basic

import requests

url = "https://datasphere.astrodynamic.space/api/item/v1/{package name}"

response = request.get(url, headers={"Authorization":"Token {api token}"})
stac_item = response.json()