API

The section shows information about the API relevant for the developers.

Extractors

eodm.extract.extract_stac_api_items(url, collections=None, bbox=None, datetime_interval=None, limit=None, query=None, filter=None)

Extracts STAC Items from a STAC API

Parameters:

Name	Type	Description	Default
url	str	Link to STAC API endpoint	required
collections	list[str] | None	List of collections to extract items from. Defaults to None.	None
bbox	tuple[float, float, float, float] | None	Bounding box to search. Defaults to None.	None
datetime_interval	str | None	Datetime interval to search. Defaults to None.	None
limit	int	Limit query to given number. Defaults to 10.	None
query	optional	STACAPI Query extension	None
filter	optional	STACAPI CQL Filter extension	None

Yields:

Type	Description
Item	Iterator[Item]: pystac Items

Source code in src/eodm/extract.py

def extract_stac_api_items(
    url: str,
    collections: list[str] | None = None,
    bbox: tuple[float, float, float, float] | None = None,
    datetime_interval: str | None = None,
    limit: int | None = None,
    query: dict | None = None,
    filter: dict | None = None,
) -> Iterator[Item]:
    """Extracts STAC Items from a STAC API

    Args:
        url (str): Link to STAC API endpoint
        collections (list[str] | None, optional): List of collections to extract items from. Defaults to None.
        bbox (tuple[float, float, float, float] | None, optional): Bounding box to search. Defaults to None.
        datetime_interval (str | None, optional): Datetime interval to search. Defaults to None.
        limit (int, optional): Limit query to given number. Defaults to 10.
        query (optional): STACAPI Query extension
        filter (optional): STACAPI CQL Filter extension

    Yields:
        Iterator[Item]: pystac Items
    """

    client = pystac_client.Client.open(url)

    search = client.search(
        collections=collections,
        bbox=bbox,
        datetime=datetime_interval,
        limit=limit,
        query=query,
        filter=filter,
    )

    yield from search.item_collection()

eodm.extract.extract_stac_api_collections(url)

Extracts STAC Collections from a STAC API

Parameters:

Name	Type	Description	Default
url	str	Link to STAC API endpoint	required

Yields:

Type	Description
Collection	Iterator[Collection]: pystac Collections

Source code in src/eodm/extract.py

def extract_stac_api_collections(url: str) -> Iterator[Collection]:
    """Extracts STAC Collections from a STAC API

    Args:
        url (str): Link to STAC API endpoint

    Yields:
        Iterator[Collection]: pystac Collections
    """

    client = pystac_client.Client.open(url)
    yield from client.get_collections()

eodm.extract.extract_opensearch_features(url, product_types, limit=0)

Extracts OpenSearch Features from an OpenSearch API

Parameters:

Name	Type	Description	Default
url	str	Link to OpenSearch API endpoint	required
productTypes	list[str]	List of productTypes to search for	required

Yields:

Type	Description
OpenSearchFeature	Iterator[OpenSearchFeature]: OpenSearch Features

Source code in src/eodm/extract.py

def extract_opensearch_features(
    url: str,
    product_types: list[str],
    limit: int = 0,
) -> Iterator[OpenSearchFeature]:
    """Extracts OpenSearch Features from an OpenSearch API

    Args:
        url (str): Link to OpenSearch API endpoint
        productTypes (list[str]): List of productTypes to search for

    Yields:
        Iterator[OpenSearchFeature]: OpenSearch Features
    """
    client = OpenSearchClient(url)

    query = {}

    # TODO: create mapper to map to STAC items
    for product_type in product_types:
        query["{eo:productType}"] = product_type
        for i, feature in enumerate(client.search(query), start=1):
            if limit and i >= limit:
                break
            yield feature

Loaders

eodm.load.load_stac_api_items(url, items, headers=None, verify=True, update=False, skip_existing=False)

Load multiple items into a STAC API

Parameters:

Name	Type	Description	Default
url	str	STAC API url	required
items	Iterable[Item]	A collection of STAC Items	required
headers	dict[str, str] | None	Headers to add to the request. Defaults to None.	None
verify	bool	Verify SSL request. Defaults to True.	True
update	bool	Update STAC Item with new content. Defaults to False.	False
skip_existing	bool	Skip Item if exists. Defaults to False.	False

Source code in src/eodm/load.py

def load_stac_api_items(
    url: str,
    items: Iterable[Item],
    headers: dict[str, str] | None = None,
    verify: bool = True,
    update: bool = False,
    skip_existing: bool = False,
) -> Iterable[Item]:
    """Load multiple items into a STAC API

    Args:
        url (str): STAC API url
        items (Iterable[Item]): A collection of STAC Items
        headers (dict[str, str] | None, optional): Headers to add to the request. Defaults to None.
        verify (bool, optional): Verify SSL request. Defaults to True.
        update (bool, optional): Update STAC Item with new content. Defaults to False.
        skip_existing (bool, optional): Skip Item if exists. Defaults to False.
    """
    if not headers:
        headers = DEFAULT_HEADERS

    for item in items:
        collection_id = item.collection_id
        items_endpoint = f"{url}/collections/{collection_id}/items"
        response = httpx.post(
            items_endpoint,
            json=item.to_dict(),
            headers=headers,
            verify=verify,
        )
        if response.status_code == 409:
            if update:
                item_endpoint = f"{items_endpoint}/{item.id}"
                response = httpx.put(
                    item_endpoint, json=item.to_dict(), headers=headers, verify=verify
                )
            if skip_existing:
                continue
        response.raise_for_status()
        yield item

eodm.load.load_stac_api_collections(url, collections, headers=None, verify=True, update=False, skip_existing=False)

Load multiple collections to a stac API

Parameters:

Name	Type	Description	Default
url	str	STAC API URL	required
collections	Iterable[Collection]	A collection of STAC Collections	required
headers	dict[str, str] | None	Additional headers to send. Defaults to None.	None
verify	bool	Verify TLS request. Defaults to True.	True
update	bool	Update the destination Collections. Defaults to False.	False
skip_existing	bool	Skip existing Collections. Defaults to False.	False

Returns:

Type	Description
Iterable[Collection]	Iterable[Collection]:

Source code in src/eodm/load.py

def load_stac_api_collections(
    url: str,
    collections: Iterable[Collection],
    headers: dict[str, str] | None = None,
    verify: bool = True,
    update: bool = False,
    skip_existing: bool = False,
) -> Iterable[Collection]:
    """Load multiple collections to a stac API

    Args:
        url (str): STAC API URL
        collections (Iterable[Collection]): A collection of STAC Collections
        headers (dict[str, str] | None, optional): Additional headers to send. Defaults to None.
        verify (bool, optional): Verify TLS request. Defaults to True.
        update (bool, optional): Update the destination Collections. Defaults to False.
        skip_existing (bool, optional): Skip existing Collections. Defaults to False.

    Returns:
        Iterable[Collection]:
    """

    if not headers:
        headers = DEFAULT_HEADERS

    collections_endpoint = f"{url}/collections"
    for collection in collections:
        response = httpx.post(
            collections_endpoint,
            json=collection.to_dict(),
            headers=headers,
            verify=verify,
        )
        if response.status_code == 409:
            if update:
                collection_endpoint = f"{collections_endpoint}/{collection.id}"
                response = httpx.put(
                    collection_endpoint,
                    json=collection.to_dict(),
                    headers=headers,
                    verify=verify,
                )
            if skip_existing:
                continue

        response.raise_for_status()
        yield collection

Opensearch

eodm.opensearch.OpenSearchClient(describe_url)

Source code in src/eodm/opensearch.py

def __init__(self, describe_url: str) -> None:
    self.describe_url = describe_url

NS = {'': 'http://www.w3.org/2005/Atom', 'opensearch': 'http://a9.com/-/spec/opensearch/1.1/', 'parameters': 'http://a9.com/-/spec/opensearch/extensions/parameters/1.0/', 'georss': 'http://www.georss.org/georss/', 'media': 'http://search.yahoo.com/mrss/', 'owc': 'http://www.opengis.net/owc/1.0/', 'eo': 'http://a9.com/-/opensearch/extensions/eo/1.0/', 'geo': 'http://a9.com/-/opensearch/extensions/geo/1.0/', 'time': 'http://a9.com/-/opensearch/extensions/time/1.0/', 'cql': 'http://a9.com/-/opensearch/extensions/cql/1.0/', 'dc': 'http://purl.org/dc/elements/1.1/'} class-attribute instance-attribute

describe_url = describe_url instance-attribute

metadata: OpenSearchMetadata cached property

Service level metadata

query_urls: list[OpenSearchUrl] cached property

List of OpenSearchUrl query objects. Can be used for constructing a query

search(query, mimetype='application/json')

Search for features in the OpenSearch service.

Parameters:

Name	Type	Description	Default
query	dict[str, Any]	Query parameters to search for. The keys should match the parameter names in the OpenSearch service and have the format {namespace}:{parameterName}. Example: {"eo:cloudCover": 10}	required
mimetype	str	Mimetype.	'application/json'

Yields:

Type	Description
OpenSearchFeature	Iterator[OpenSearchFeature]: Iterator of OpenSearchFeature object

Source code in src/eodm/opensearch.py

def search(
    self,
    query: dict[str, Any],
    mimetype: str = "application/json",
) -> Iterator[OpenSearchFeature]:
    """Search for features in the OpenSearch service.

    Args:
        query (dict[str, Any]): Query parameters to search for.
            The keys should match the parameter names in the OpenSearch service and
            have the format `{namespace}:{parameterName}`.
            Example: {"eo:cloudCover": 10}
        mimetype (str, optional): Mimetype.

    Yields:
        Iterator[OpenSearchFeature]: Iterator of OpenSearchFeature object
    """
    search_url = self._prepare_search_url(mimetype, query)
    next_page = True

    while next_page:
        with Client(transport=DEFAULT_TRANSPORT) as client:
            response = client.get(search_url)
            response.raise_for_status()

        # here branch logic for different mimetypes, only json so far known
        feature_collection = OpenSearchResults.model_validate(response.json())

        yield from feature_collection.features

        next_page, url = self._get_next_page(feature_collection.properties.links)

eodm.opensearch.OpenSearchFeature

Bases: OpenSearchFeatureType

to_dict()

Converts to a dictionary

Source code in src/eodm/opensearch.py

def to_dict(self) -> dict:
    """Converts to a dictionary"""
    return self.model_dump()

eodm.opensearch.OpenSearchUrl

Bases: BaseModel

parameters: dict[str, OpenSearchParameter] instance-attribute

template_url: str instance-attribute

type: str instance-attribute

Serializers

eodm.serializers

Mappable

Bases: Protocol

to_dict()

Source code in src/eodm/serializers.py

def to_dict(self): ...

default_serializer(items)

Serializes a list of Mappables (implementing to_dict()) to json strings individually

Parameters:

Name	Type	Description	Default
items	Iterable[Mappable]	A collection of Mappable items	required

Returns:

Type	Description
Iterable[str]	Iterable[str]: item as a string

Yields:

Type	Description
Iterable[str]	Iterator[Iterable[str]]: Collection of json strings

Source code in src/eodm/serializers.py

def default_serializer(items: Iterable[Mappable]) -> Iterable[str]:
    """Serializes a list of Mappables (implementing to_dict()) to json strings
    individually


    Args:
        items (Iterable[Mappable]): A collection of Mappable items

    Returns:
        Iterable[str]: item as a string

    Yields:
        Iterator[Iterable[str]]: Collection of json strings
    """

    for item in items:
        yield json.dumps(item.to_dict(), default=str)

json_serializer(items)

Serializes a list of Mappables (implementing to_dict()) to a json list

Parameters:

Name	Type	Description	Default
items	Iterable[Mappable]	A collection of Mappable items	required

Returns:

Name	Type	Description
str	str	items as a json list

Source code in src/eodm/serializers.py

def json_serializer(items: Iterable[Mappable]) -> str:
    """Serializes a list of Mappables (implementing to_dict()) to a json list

    Args:
        items (Iterable[Mappable]): A collection of Mappable items

    Returns:
        str: items as a json list
    """
    return json.dumps([item.to_dict() for item in items], default=str)

STAC Contributions

eodm.stac_contrib

FSSpecStacIO

Bases: StacIO

Extension of StacIO to allow working with different filesystems in STAC using fsspec.

More information: https://pystac.readthedocs.io/en/stable/concepts.html#i-o-in-pystac

read_text(source, *args, **kwargs)

Source code in src/eodm/stac_contrib.py

def read_text(self, source: HREF, *args: Any, **kwargs: Any) -> str:
    if fs := kwargs.get("filesystem"):
        with fs.open(source, "r", *args, **kwargs) as f:
            return f.read()
    else:
        with fsspec.open(source, "r", *args, **kwargs) as f:
            return f.read()

write_text(dest, txt, *args, **kwargs)

Source code in src/eodm/stac_contrib.py

def write_text(self, dest: HREF, txt: str, *args: Any, **kwargs: Any) -> None:
    if fs := kwargs.get("filesystem"):
        with fs.open(dest, "w", *args, **kwargs) as f:
            f.write(txt)
    else:
        with fsspec.open(dest, "w", *args, **kwargs) as f:
            f.write(txt)