Module max_ard.base_collections

Base collection types for ARD

Provides

  • Pydantic Models that are used in multiple places
  • Submitted and Succeeded ABCs that add "requires" decorators
  • Collections of tiles
  • Collections of tiles from a common acquisition
  • Collections of tiles covering a common grid cells

These collection objects assume a generic concept of a "tile". They can be used to group tiles-to-be found in a Select, or real tiles delivered to a storage location.

Expand source code 
"""Base collection types for ARD

Provides
--------
- Pydantic Models that are used in multiple places
- Submitted and Succeeded ABCs that add "requires" decorators
- Collections of tiles
- Collections of tiles from a common acquisition
- Collections of tiles covering a common grid cells

These collection objects assume a generic concept of a "tile". They can be used to
group tiles-to-be found in a Select, or real tiles delivered to a storage location.
"""
from abc import ABC, abstractmethod
from functools import wraps
from tempfile import NamedTemporaryFile
from typing import Any, List, Union

from maxar_ard_grid.grid import Cell
from pydantic import BaseModel
from shapely.geometry import MultiPolygon, mapping

from max_ard.dependency_support import HAS_RASTERIO
from max_ard.exceptions import NotFinished, NotSubmitted, NotSuccessful, SelectError
from max_ard.processing import AcquisitionReader

if HAS_RASTERIO:
    import rasterio


def hydrate(cls, things, key=None):
    """Convenience function for hydrating objects from a list of dicts

    Arguments
    ---------
    cls: Object class to create
    things: iterable of raw object dicts
    key: like in sort(), function that exposes where in the dict the object's arguments are

    Returns
    -------
    list
        list of objects of type cls

    Notes
    -----
    If a key function is not provided, the dict is assumed to hold the object arguments.
    Operator.itemgetter(<name>) or a lamba x: x[<name>] can be used.

    """
    if key is None:
        key = lambda x: x
    return [cls(**key(thing)) for thing in things]


def hydrate_with_responses(cls, response_cls, responses, **kwargs):
    instances = []
    for response in responses:
        instance = cls(**kwargs)
        instance.response = response_cls(**response)
        instances.append(instance)

    return instances


###
# Common Pydantic Models
###


class ARDModel(BaseModel):
    """A Pydantic BaseModel with sanitized dicts

    The ARD API can be picky about Nones and empty objects so for request payloads
    we can use the to_payload method to generate a cleaned up payload"""

    def to_payload(self):
        """sends only parameters with values"""
        params = self.dict().copy()
        payload = {k: v for k, v in params.items() if v not in (None, (), [])}
        return payload


class EmailNotification(BaseModel):
    """A model for Email Notifications"""

    type: str = "email"
    address: str

    class Config:
        allow_mutation: False


class SNSNotification(BaseModel):
    """A model for SNS Notifications"""

    type: str = "sns"
    topic_arn: str

    class Config:
        allow_mutation: False


class UsageArea(BaseModel):
    """A model to hold ARD usage by area"""

    fresh_imagery_sqkm: float
    standard_imagery_sqkm: float
    training_imagery_sqkm: float
    tasking_imagery_sqkm: Any
    total_imagery_sqkm: float
    estimate: bool


class UsageCost(BaseModel):
    """A model to hold ARD image costs, estimated or consumed

    A value of None for a cost means your account does not have pricing set
    for this category. Without set pricing this category can not be ordered."""

    fresh_imagery_cost: Union[float, None]
    standard_imagery_cost: Union[float, None]
    training_imagery_cost: Union[float, None]
    tasking_imagery_cost: Any
    total_imagery_cost: Union[float, None]
    estimate: bool


class UsageLimits(BaseModel):
    """A model to hold account limits"""

    # limit_sqkm: float # dropped Jan 20
    fresh_imagery_fee_limit: float
    standard_imagery_fee_limit: float
    training_imagery_fee_limit: float
    tasking_imagery_fee_limit: Any
    annual_subscription_fee_limit: float

    class Config:
        extra = "allow"


class UsageAvailable(BaseModel):
    """A model to hold account balances

    When returned with a select, these numbers reflect the account balance at the time
    the select was submitted. Usage estimates are not applied to these numbers."""

    # available_sqkm: float # dropped Jan 20
    fresh_imagery_balance: float
    standard_imagery_balance: float
    training_imagery_balance: float
    tasking_imagery_balance: Any
    total_imagery_balance: float

    class Config:
        extra = "allow"


###
# Validation ABCs
###


class Submitted(ABC):
    """An ABC for objects that can be submitted

    - Concrete classes that inherit from Submitted must override submitted()
    - `submitted` is usually exposed as a @property
    - how you determine an object has been submitted is up to you

    For methods that require the object to have been submitted, wrap it with
    the @Submitted.required decorator
    """

    @abstractmethod
    def submitted(self):
        pass

    @staticmethod
    def required(func):
        @wraps(func)
        def func_wrapper(instance, *args, **kwargs):
            if not instance.submitted:
                raise NotSubmitted(
                    f"Can't access `{func.__name__}` until {type(instance).__name__} has been submitted"
                )
            res = func(instance, *args, **kwargs)
            return res

        return func_wrapper


class Succeeded(ABC):
    """An ABC for objects that have a succeeded state, and by definition
    must also be submitted in order to run

    - Concrete classes that inherit from Submitted must override
        - submitted()
        - finished()
        - succeeded()
    - see also Submitted

    For methods that require the object to have succeeded, wrap it with
    the @Succeeded.required decorator
    """

    @abstractmethod
    def submitted(self):
        pass

    @abstractmethod
    def finished(self):
        pass

    @abstractmethod
    def succeeded(self):
        pass

    @staticmethod
    def required(func):
        @wraps(func)
        def func_wrapper(instance, *args, **kwargs):
            if not instance.submitted:
                raise NotSubmitted(
                    f"Can't access `{func.__name__}` until {type(instance).__name__} has been submitted"
                )
            if not instance.finished:
                raise NotFinished(
                    f"Can't access `{func.__name__}` until {type(instance).__name__} has finished running"
                )
            if not instance.succeeded:
                # This error should probably be generic, but only Selects use this ABC so far
                if type(instance).__name__ == "Select":
                    raise SelectError(f"Can't access `{func.__name__}` due to an error")
                else:
                    raise NotSuccessful(f"Can't access `{func.__name__}` due to an error")

            res = func(instance, *args, **kwargs)
            return res

        return func_wrapper


###
# Classes for ARD Concepts
###


class Acquisition(list):
    """Container for one or more tiles from a single acquisition

    These are generated by ARDCollections and would not normally be initialized"""

    def __init__(self, acq_id: str, tiles: list = []) -> None:

        self.acq_id = acq_id
        """ str: ID of this acquisition """
        super().__init__(*tiles)

    @property
    def cells(self) -> List[Cell]:
        cells = set([tile.cell for tile in self])
        return list(cells)

    def get_tile_from_cell(self, cell):
        cell = Cell(cell)
        for tile in self:
            if tile.cell == cell:
                return tile
        raise ValueError("Tile not found in this acquisition")

    def __str__(self):
        return f"<Acquisition at {self.acq_id} ({len(self)} tiles)>"

    def __repr__(self):
        tiles = ", ".join([f"<{type(tile).__name__} at {tile.cell.id}>" for tile in self])
        return f"<Acquisition of {self.acq_id} [{tiles}]>"

    def __hash__(self):
        """Hash of cell ID"""
        return hash(self.acq_id)

    def open_acquisition(self):
        """Return a Rasterio reader that accesses all of the Acquisition object's tiles.

        A Rasterio DatasetReader opened on a VRT of all acquisition tiles"""

        return AcquisitionReader(self)

    @property
    def date(self):
        return self[0].date

    @property
    def properties(self) -> dict:
        """Strip level properties"""

        props = self[0].properties
        # todo: check if catalog_id is still passed in metadata
        keep = ["catalog_id", "acquisition_id", "platform", "scan_direction", "epsg", "datetime"]
        strip_props = {}
        for key in keep:
            try:
                strip_props[key] = props[key]
            except KeyError:
                pass
        return strip_props

    @property
    def __geo_interface__(self) -> dict:
        geoms = [tile.cell.geom_WGS84 for tile in self]
        return mapping(MultiPolygon(geoms))


class Stack(list):
    """Container for one or more tiles in an ARD grid cell

    These are generated by ARDCollections and would not normally be initialized"""

    def __init__(self, cell: Cell, tiles: list = []) -> None:

        self.cell = Cell(cell)
        super().__init__(*tiles)

    @property
    def tiles(self):
        """not needed but provides access to tiles for GeoMixin"""
        return self

    def get_tile_from_acquisition(self, acq_id):
        for tile in self:
            if tile.acq_id == acq_id:
                return tile
        raise ValueError("Tile not found in this stack")

    @property
    def acquisition_ids(self) -> List[str]:
        ids = set([tile.acq_id for tile in self])
        return list(ids)

    def __str__(self):
        return f"<Stack at {self.cell.id} ({len(self)} tiles)>"

    def __repr__(self):
        tiles = ", ".join([f"<{type(tile).__name__} of {tile.acq_id}>" for tile in self])
        return f"<Stack at {self.cell.id} [{tiles}]>"

    def __hash__(self):
        """Hash of cell ID"""
        return hash(self.cell)

    @property
    def __geo_interface__(self) -> dict:
        geoms = [tile.cell.geom_WGS84 for tile in self]
        return mapping(MultiPolygon(geoms))


class Store(dict):
    """A defaultdict-like k:v store for objects"""

    def __init__(self, object_type, *args, **kwargs):
        self.object_type = object_type
        super().__init__(*args, **kwargs)

    def __missing__(self, key):
        res = self[key] = self.object_type(key)
        return res


class BaseCollection:
    def __init__(self):
        self._reset()

    def _reset(self):
        """Used by ARDCollection when a rescan is triggered"""
        self._acquisitions = Store(Acquisition)
        self._stacks = Store(Stack)
        self._zones = set()
        self._dates = set()

    def add_tile(self, tile):
        self._acquisitions[tile.acq_id].append(tile)
        self._stacks[tile.cell].append(tile)
        self._zones.add(tile.cell.zone)
        self._dates.add(tile.date)

    @property
    def tiles(self):
        """All the ARD Tiles in this collection"""
        return [tile for tiles in self.stacks for tile in tiles]

    def get_tile(self, acq_id, cell):
        """Get a tile from the collection

        Parameters
        ----------
        acq_id : str
            Acquisition ID
        cell : str or maxar_ard_grid.Cell
            Cell of the tile, can be a cell ID or Cell object

        Returns
        -------
        ARDTile or SelectTile
        """
        cell = Cell(cell)
        acq = self.get_acquisition(acq_id)
        for tile in acq:
            if tile.cell == cell:
                return tile
        raise ValueError("Tile not found in this collection")

    @property
    def dates(self) -> List[str]:
        """All of the dates of acquisitions in the ARDCollection"""
        dates = list(self._dates)
        dates.sort()
        return dates

    @property
    def start_date(self) -> str:
        """Earliest date of acquisitions in the ARDCollection as YYYY-MM-DD"""
        return self.dates[0]

    @property
    def end_date(self) -> str:
        # does the type get caught?
        """Latest date of acquisitions in the ARDCollection as YYYY-MM-DD"""
        return self.dates[-1]

    @property
    def zones(self) -> List[int]:
        """The UTM zones covered by tiles in the ARDCollection"""
        return list(self._zones)

    #
    # Stacks
    #

    @property
    def stacks(self):
        """All of ARD grid stacks in the collection"""
        return list(self._stacks.values())

    @property
    def cells(self):
        """All of the ARD grid Cell objects in the collection"""
        return list(self._stacks.keys())

    def get_stack(self, cell):
        """Get a stack of tiles from the collection

        A Stack object is a container of all tiles covering a given ARD grid cell.
        Parameters
        ----------
        cell : str or maxar_ard_grid.Cell
            Cell of the tile, can be a cell ID or Cell object

        Returns
        -------
        Stack
        """
        cell = Cell(cell)
        try:
            return self._stacks[cell]
        except KeyError:
            raise ValueError(f"Stack for cell {cell} not found in this collection")

    #
    # Acquisitions
    #

    @property
    def acquisitions(self):
        """All of the acquisitions present in the collection"""
        return list(self._acquisitions.values())

    @property
    def acquisition_ids(self):
        """All of the acquisitions IDs present in the collection"""
        return list(self._acquisitions.keys())

    @property
    def acq_ids(self):
        """Shorthand name for .acquisition_ids"""
        return self.acquisition_ids

    def get_acquisition(self, acq_id):
        """Get an acquisition from the collection

        An Acquisition object is a container of all tiles from a given acquisition.

        Parameters
        ----------
        acq_id : str
            Acquisition ID

        Returns
        -------
        Acquisition
        """
        try:
            return self._acquisitions[acq_id]
        except KeyError:
            raise ValueError(f"Aquisition for ID {acq_id} not found in this collection")

    def as_order(self) -> List:
        """Return the collection in the ARD Order format"""

        order = []
        for acq in self.acquisitions:
            order.append({"id": acq.acq_id, "cells": [cell.id for cell in acq.cells]})
        return order

    @property
    def __geo_interface__(self) -> dict:
        geoms = [tile.cell.geom_WGS84 for tile in self.tiles]
        return mapping(MultiPolygon(geoms))

Functions

def hydrate(cls, things, key=None)

Convenience function for hydrating objects from a list of dicts

Arguments

cls : Object class to create
 
things : iterable of raw object dicts
 
key : like in sort(), function that exposes where in the dict the object's arguments are
 

Returns

list
list of objects of type cls

Notes

If a key function is not provided, the dict is assumed to hold the object arguments. Operator.itemgetter() or a lamba x: x[] can be used.

Expand source code 
def hydrate(cls, things, key=None):
    """Convenience function for hydrating objects from a list of dicts

    Arguments
    ---------
    cls: Object class to create
    things: iterable of raw object dicts
    key: like in sort(), function that exposes where in the dict the object's arguments are

    Returns
    -------
    list
        list of objects of type cls

    Notes
    -----
    If a key function is not provided, the dict is assumed to hold the object arguments.
    Operator.itemgetter(<name>) or a lamba x: x[<name>] can be used.

    """
    if key is None:
        key = lambda x: x
    return [cls(**key(thing)) for thing in things]
def hydrate_with_responses(cls, response_cls, responses, **kwargs)
Expand source code 
def hydrate_with_responses(cls, response_cls, responses, **kwargs):
    instances = []
    for response in responses:
        instance = cls(**kwargs)
        instance.response = response_cls(**response)
        instances.append(instance)

    return instances

Classes

class ARDModel (**data: Any)

A Pydantic BaseModel with sanitized dicts

The ARD API can be picky about Nones and empty objects so for request payloads we can use the to_payload method to generate a cleaned up payload

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class ARDModel(BaseModel):
    """A Pydantic BaseModel with sanitized dicts

    The ARD API can be picky about Nones and empty objects so for request payloads
    we can use the to_payload method to generate a cleaned up payload"""

    def to_payload(self):
        """sends only parameters with values"""
        params = self.dict().copy()
        payload = {k: v for k, v in params.items() if v not in (None, (), [])}
        return payload

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Subclasses

  • MonitorRequest
  • max_ard.order.OrderRequest
  • max_ard.select.SelectRequest

Methods

def to_payload(self)

sends only parameters with values

Expand source code 
def to_payload(self):
    """sends only parameters with values"""
    params = self.dict().copy()
    payload = {k: v for k, v in params.items() if v not in (None, (), [])}
    return payload
class Acquisition (acq_id: str, tiles: list = [])

Container for one or more tiles from a single acquisition

These are generated by ARDCollections and would not normally be initialized

Expand source code 
class Acquisition(list):
    """Container for one or more tiles from a single acquisition

    These are generated by ARDCollections and would not normally be initialized"""

    def __init__(self, acq_id: str, tiles: list = []) -> None:

        self.acq_id = acq_id
        """ str: ID of this acquisition """
        super().__init__(*tiles)

    @property
    def cells(self) -> List[Cell]:
        cells = set([tile.cell for tile in self])
        return list(cells)

    def get_tile_from_cell(self, cell):
        cell = Cell(cell)
        for tile in self:
            if tile.cell == cell:
                return tile
        raise ValueError("Tile not found in this acquisition")

    def __str__(self):
        return f"<Acquisition at {self.acq_id} ({len(self)} tiles)>"

    def __repr__(self):
        tiles = ", ".join([f"<{type(tile).__name__} at {tile.cell.id}>" for tile in self])
        return f"<Acquisition of {self.acq_id} [{tiles}]>"

    def __hash__(self):
        """Hash of cell ID"""
        return hash(self.acq_id)

    def open_acquisition(self):
        """Return a Rasterio reader that accesses all of the Acquisition object's tiles.

        A Rasterio DatasetReader opened on a VRT of all acquisition tiles"""

        return AcquisitionReader(self)

    @property
    def date(self):
        return self[0].date

    @property
    def properties(self) -> dict:
        """Strip level properties"""

        props = self[0].properties
        # todo: check if catalog_id is still passed in metadata
        keep = ["catalog_id", "acquisition_id", "platform", "scan_direction", "epsg", "datetime"]
        strip_props = {}
        for key in keep:
            try:
                strip_props[key] = props[key]
            except KeyError:
                pass
        return strip_props

    @property
    def __geo_interface__(self) -> dict:
        geoms = [tile.cell.geom_WGS84 for tile in self]
        return mapping(MultiPolygon(geoms))

Ancestors

  • builtins.list

Instance variables

var acq_id

str: ID of this acquisition

var cells : List[Cell]
Expand source code 
@property
def cells(self) -> List[Cell]:
    cells = set([tile.cell for tile in self])
    return list(cells)
var date
Expand source code 
@property
def date(self):
    return self[0].date
var properties : dict

Strip level properties

Expand source code 
@property
def properties(self) -> dict:
    """Strip level properties"""

    props = self[0].properties
    # todo: check if catalog_id is still passed in metadata
    keep = ["catalog_id", "acquisition_id", "platform", "scan_direction", "epsg", "datetime"]
    strip_props = {}
    for key in keep:
        try:
            strip_props[key] = props[key]
        except KeyError:
            pass
    return strip_props

Methods

def get_tile_from_cell(self, cell)
Expand source code 
def get_tile_from_cell(self, cell):
    cell = Cell(cell)
    for tile in self:
        if tile.cell == cell:
            return tile
    raise ValueError("Tile not found in this acquisition")
def open_acquisition(self)

Return a Rasterio reader that accesses all of the Acquisition object's tiles.

A Rasterio DatasetReader opened on a VRT of all acquisition tiles

Expand source code 
def open_acquisition(self):
    """Return a Rasterio reader that accesses all of the Acquisition object's tiles.

    A Rasterio DatasetReader opened on a VRT of all acquisition tiles"""

    return AcquisitionReader(self)
class BaseCollection
Expand source code 
class BaseCollection:
    def __init__(self):
        self._reset()

    def _reset(self):
        """Used by ARDCollection when a rescan is triggered"""
        self._acquisitions = Store(Acquisition)
        self._stacks = Store(Stack)
        self._zones = set()
        self._dates = set()

    def add_tile(self, tile):
        self._acquisitions[tile.acq_id].append(tile)
        self._stacks[tile.cell].append(tile)
        self._zones.add(tile.cell.zone)
        self._dates.add(tile.date)

    @property
    def tiles(self):
        """All the ARD Tiles in this collection"""
        return [tile for tiles in self.stacks for tile in tiles]

    def get_tile(self, acq_id, cell):
        """Get a tile from the collection

        Parameters
        ----------
        acq_id : str
            Acquisition ID
        cell : str or maxar_ard_grid.Cell
            Cell of the tile, can be a cell ID or Cell object

        Returns
        -------
        ARDTile or SelectTile
        """
        cell = Cell(cell)
        acq = self.get_acquisition(acq_id)
        for tile in acq:
            if tile.cell == cell:
                return tile
        raise ValueError("Tile not found in this collection")

    @property
    def dates(self) -> List[str]:
        """All of the dates of acquisitions in the ARDCollection"""
        dates = list(self._dates)
        dates.sort()
        return dates

    @property
    def start_date(self) -> str:
        """Earliest date of acquisitions in the ARDCollection as YYYY-MM-DD"""
        return self.dates[0]

    @property
    def end_date(self) -> str:
        # does the type get caught?
        """Latest date of acquisitions in the ARDCollection as YYYY-MM-DD"""
        return self.dates[-1]

    @property
    def zones(self) -> List[int]:
        """The UTM zones covered by tiles in the ARDCollection"""
        return list(self._zones)

    #
    # Stacks
    #

    @property
    def stacks(self):
        """All of ARD grid stacks in the collection"""
        return list(self._stacks.values())

    @property
    def cells(self):
        """All of the ARD grid Cell objects in the collection"""
        return list(self._stacks.keys())

    def get_stack(self, cell):
        """Get a stack of tiles from the collection

        A Stack object is a container of all tiles covering a given ARD grid cell.
        Parameters
        ----------
        cell : str or maxar_ard_grid.Cell
            Cell of the tile, can be a cell ID or Cell object

        Returns
        -------
        Stack
        """
        cell = Cell(cell)
        try:
            return self._stacks[cell]
        except KeyError:
            raise ValueError(f"Stack for cell {cell} not found in this collection")

    #
    # Acquisitions
    #

    @property
    def acquisitions(self):
        """All of the acquisitions present in the collection"""
        return list(self._acquisitions.values())

    @property
    def acquisition_ids(self):
        """All of the acquisitions IDs present in the collection"""
        return list(self._acquisitions.keys())

    @property
    def acq_ids(self):
        """Shorthand name for .acquisition_ids"""
        return self.acquisition_ids

    def get_acquisition(self, acq_id):
        """Get an acquisition from the collection

        An Acquisition object is a container of all tiles from a given acquisition.

        Parameters
        ----------
        acq_id : str
            Acquisition ID

        Returns
        -------
        Acquisition
        """
        try:
            return self._acquisitions[acq_id]
        except KeyError:
            raise ValueError(f"Aquisition for ID {acq_id} not found in this collection")

    def as_order(self) -> List:
        """Return the collection in the ARD Order format"""

        order = []
        for acq in self.acquisitions:
            order.append({"id": acq.acq_id, "cells": [cell.id for cell in acq.cells]})
        return order

    @property
    def __geo_interface__(self) -> dict:
        geoms = [tile.cell.geom_WGS84 for tile in self.tiles]
        return mapping(MultiPolygon(geoms))

Subclasses

Instance variables

var acq_ids

Shorthand name for .acquisition_ids

Expand source code 
@property
def acq_ids(self):
    """Shorthand name for .acquisition_ids"""
    return self.acquisition_ids
var acquisition_ids

All of the acquisitions IDs present in the collection

Expand source code 
@property
def acquisition_ids(self):
    """All of the acquisitions IDs present in the collection"""
    return list(self._acquisitions.keys())
var acquisitions

All of the acquisitions present in the collection

Expand source code 
@property
def acquisitions(self):
    """All of the acquisitions present in the collection"""
    return list(self._acquisitions.values())
var cells

All of the ARD grid Cell objects in the collection

Expand source code 
@property
def cells(self):
    """All of the ARD grid Cell objects in the collection"""
    return list(self._stacks.keys())
var dates : List[str]

All of the dates of acquisitions in the ARDCollection

Expand source code 
@property
def dates(self) -> List[str]:
    """All of the dates of acquisitions in the ARDCollection"""
    dates = list(self._dates)
    dates.sort()
    return dates
var end_date : str

Latest date of acquisitions in the ARDCollection as YYYY-MM-DD

Expand source code 
@property
def end_date(self) -> str:
    # does the type get caught?
    """Latest date of acquisitions in the ARDCollection as YYYY-MM-DD"""
    return self.dates[-1]
var stacks

All of ARD grid stacks in the collection

Expand source code 
@property
def stacks(self):
    """All of ARD grid stacks in the collection"""
    return list(self._stacks.values())
var start_date : str

Earliest date of acquisitions in the ARDCollection as YYYY-MM-DD

Expand source code 
@property
def start_date(self) -> str:
    """Earliest date of acquisitions in the ARDCollection as YYYY-MM-DD"""
    return self.dates[0]
var tiles

All the ARD Tiles in this collection

Expand source code 
@property
def tiles(self):
    """All the ARD Tiles in this collection"""
    return [tile for tiles in self.stacks for tile in tiles]
var zones : List[int]

The UTM zones covered by tiles in the ARDCollection

Expand source code 
@property
def zones(self) -> List[int]:
    """The UTM zones covered by tiles in the ARDCollection"""
    return list(self._zones)

Methods

def add_tile(self, tile)
Expand source code 
def add_tile(self, tile):
    self._acquisitions[tile.acq_id].append(tile)
    self._stacks[tile.cell].append(tile)
    self._zones.add(tile.cell.zone)
    self._dates.add(tile.date)
def as_order(self) ‑> List[~T]

Return the collection in the ARD Order format

Expand source code 
def as_order(self) -> List:
    """Return the collection in the ARD Order format"""

    order = []
    for acq in self.acquisitions:
        order.append({"id": acq.acq_id, "cells": [cell.id for cell in acq.cells]})
    return order
def get_acquisition(self, acq_id)

Get an acquisition from the collection

An Acquisition object is a container of all tiles from a given acquisition.

Parameters

acq_id : str
Acquisition ID

Returns

Acquisition
 
Expand source code 
def get_acquisition(self, acq_id):
    """Get an acquisition from the collection

    An Acquisition object is a container of all tiles from a given acquisition.

    Parameters
    ----------
    acq_id : str
        Acquisition ID

    Returns
    -------
    Acquisition
    """
    try:
        return self._acquisitions[acq_id]
    except KeyError:
        raise ValueError(f"Aquisition for ID {acq_id} not found in this collection")
def get_stack(self, cell)

Get a stack of tiles from the collection

A Stack object is a container of all tiles covering a given ARD grid cell. Parameters

cell : str or maxar_ard_grid.Cell
Cell of the tile, can be a cell ID or Cell object

Returns

Stack
 
Expand source code 
def get_stack(self, cell):
    """Get a stack of tiles from the collection

    A Stack object is a container of all tiles covering a given ARD grid cell.
    Parameters
    ----------
    cell : str or maxar_ard_grid.Cell
        Cell of the tile, can be a cell ID or Cell object

    Returns
    -------
    Stack
    """
    cell = Cell(cell)
    try:
        return self._stacks[cell]
    except KeyError:
        raise ValueError(f"Stack for cell {cell} not found in this collection")
def get_tile(self, acq_id, cell)

Get a tile from the collection

Parameters

acq_id : str
Acquisition ID
cell : str or maxar_ard_grid.Cell
Cell of the tile, can be a cell ID or Cell object

Returns

ARDTile or SelectTile
 
Expand source code 
def get_tile(self, acq_id, cell):
    """Get a tile from the collection

    Parameters
    ----------
    acq_id : str
        Acquisition ID
    cell : str or maxar_ard_grid.Cell
        Cell of the tile, can be a cell ID or Cell object

    Returns
    -------
    ARDTile or SelectTile
    """
    cell = Cell(cell)
    acq = self.get_acquisition(acq_id)
    for tile in acq:
        if tile.cell == cell:
            return tile
    raise ValueError("Tile not found in this collection")
class EmailNotification (**data: Any)

A model for Email Notifications

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class EmailNotification(BaseModel):
    """A model for Email Notifications"""

    type: str = "email"
    address: str

    class Config:
        allow_mutation: False

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var Config
var address : str
var type : str
class SNSNotification (**data: Any)

A model for SNS Notifications

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class SNSNotification(BaseModel):
    """A model for SNS Notifications"""

    type: str = "sns"
    topic_arn: str

    class Config:
        allow_mutation: False

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var Config
var topic_arn : str
var type : str
class Stack (cell: Cell, tiles: list = [])

Container for one or more tiles in an ARD grid cell

These are generated by ARDCollections and would not normally be initialized

Expand source code 
class Stack(list):
    """Container for one or more tiles in an ARD grid cell

    These are generated by ARDCollections and would not normally be initialized"""

    def __init__(self, cell: Cell, tiles: list = []) -> None:

        self.cell = Cell(cell)
        super().__init__(*tiles)

    @property
    def tiles(self):
        """not needed but provides access to tiles for GeoMixin"""
        return self

    def get_tile_from_acquisition(self, acq_id):
        for tile in self:
            if tile.acq_id == acq_id:
                return tile
        raise ValueError("Tile not found in this stack")

    @property
    def acquisition_ids(self) -> List[str]:
        ids = set([tile.acq_id for tile in self])
        return list(ids)

    def __str__(self):
        return f"<Stack at {self.cell.id} ({len(self)} tiles)>"

    def __repr__(self):
        tiles = ", ".join([f"<{type(tile).__name__} of {tile.acq_id}>" for tile in self])
        return f"<Stack at {self.cell.id} [{tiles}]>"

    def __hash__(self):
        """Hash of cell ID"""
        return hash(self.cell)

    @property
    def __geo_interface__(self) -> dict:
        geoms = [tile.cell.geom_WGS84 for tile in self]
        return mapping(MultiPolygon(geoms))

Ancestors

  • builtins.list

Instance variables

var acquisition_ids : List[str]
Expand source code 
@property
def acquisition_ids(self) -> List[str]:
    ids = set([tile.acq_id for tile in self])
    return list(ids)
var tiles

not needed but provides access to tiles for GeoMixin

Expand source code 
@property
def tiles(self):
    """not needed but provides access to tiles for GeoMixin"""
    return self

Methods

def get_tile_from_acquisition(self, acq_id)
Expand source code 
def get_tile_from_acquisition(self, acq_id):
    for tile in self:
        if tile.acq_id == acq_id:
            return tile
    raise ValueError("Tile not found in this stack")
class Store (object_type, *args, **kwargs)

A defaultdict-like k:v store for objects

Expand source code 
class Store(dict):
    """A defaultdict-like k:v store for objects"""

    def __init__(self, object_type, *args, **kwargs):
        self.object_type = object_type
        super().__init__(*args, **kwargs)

    def __missing__(self, key):
        res = self[key] = self.object_type(key)
        return res

Ancestors

  • builtins.dict
class Submitted

An ABC for objects that can be submitted

  • Concrete classes that inherit from Submitted must override submitted()
  • submitted is usually exposed as a @property
  • how you determine an object has been submitted is up to you

For methods that require the object to have been submitted, wrap it with the @Submitted.required decorator

Expand source code 
class Submitted(ABC):
    """An ABC for objects that can be submitted

    - Concrete classes that inherit from Submitted must override submitted()
    - `submitted` is usually exposed as a @property
    - how you determine an object has been submitted is up to you

    For methods that require the object to have been submitted, wrap it with
    the @Submitted.required decorator
    """

    @abstractmethod
    def submitted(self):
        pass

    @staticmethod
    def required(func):
        @wraps(func)
        def func_wrapper(instance, *args, **kwargs):
            if not instance.submitted:
                raise NotSubmitted(
                    f"Can't access `{func.__name__}` until {type(instance).__name__} has been submitted"
                )
            res = func(instance, *args, **kwargs)
            return res

        return func_wrapper

Ancestors

  • abc.ABC

Subclasses

Static methods

def required(func)
Expand source code 
@staticmethod
def required(func):
    @wraps(func)
    def func_wrapper(instance, *args, **kwargs):
        if not instance.submitted:
            raise NotSubmitted(
                f"Can't access `{func.__name__}` until {type(instance).__name__} has been submitted"
            )
        res = func(instance, *args, **kwargs)
        return res

    return func_wrapper

Methods

def submitted(self)
Expand source code 
@abstractmethod
def submitted(self):
    pass
class Succeeded

An ABC for objects that have a succeeded state, and by definition must also be submitted in order to run

  • Concrete classes that inherit from Submitted must override
    • submitted()
    • finished()
    • succeeded()
  • see also Submitted

For methods that require the object to have succeeded, wrap it with the @Succeeded.required decorator

Expand source code 
class Succeeded(ABC):
    """An ABC for objects that have a succeeded state, and by definition
    must also be submitted in order to run

    - Concrete classes that inherit from Submitted must override
        - submitted()
        - finished()
        - succeeded()
    - see also Submitted

    For methods that require the object to have succeeded, wrap it with
    the @Succeeded.required decorator
    """

    @abstractmethod
    def submitted(self):
        pass

    @abstractmethod
    def finished(self):
        pass

    @abstractmethod
    def succeeded(self):
        pass

    @staticmethod
    def required(func):
        @wraps(func)
        def func_wrapper(instance, *args, **kwargs):
            if not instance.submitted:
                raise NotSubmitted(
                    f"Can't access `{func.__name__}` until {type(instance).__name__} has been submitted"
                )
            if not instance.finished:
                raise NotFinished(
                    f"Can't access `{func.__name__}` until {type(instance).__name__} has finished running"
                )
            if not instance.succeeded:
                # This error should probably be generic, but only Selects use this ABC so far
                if type(instance).__name__ == "Select":
                    raise SelectError(f"Can't access `{func.__name__}` due to an error")
                else:
                    raise NotSuccessful(f"Can't access `{func.__name__}` due to an error")

            res = func(instance, *args, **kwargs)
            return res

        return func_wrapper

Ancestors

  • abc.ABC

Subclasses

Static methods

def required(func)
Expand source code 
@staticmethod
def required(func):
    @wraps(func)
    def func_wrapper(instance, *args, **kwargs):
        if not instance.submitted:
            raise NotSubmitted(
                f"Can't access `{func.__name__}` until {type(instance).__name__} has been submitted"
            )
        if not instance.finished:
            raise NotFinished(
                f"Can't access `{func.__name__}` until {type(instance).__name__} has finished running"
            )
        if not instance.succeeded:
            # This error should probably be generic, but only Selects use this ABC so far
            if type(instance).__name__ == "Select":
                raise SelectError(f"Can't access `{func.__name__}` due to an error")
            else:
                raise NotSuccessful(f"Can't access `{func.__name__}` due to an error")

        res = func(instance, *args, **kwargs)
        return res

    return func_wrapper

Methods

def finished(self)
Expand source code 
@abstractmethod
def finished(self):
    pass
def submitted(self)
Expand source code 
@abstractmethod
def submitted(self):
    pass
def succeeded(self)
Expand source code 
@abstractmethod
def succeeded(self):
    pass
class UsageArea (**data: Any)

A model to hold ARD usage by area

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class UsageArea(BaseModel):
    """A model to hold ARD usage by area"""

    fresh_imagery_sqkm: float
    standard_imagery_sqkm: float
    training_imagery_sqkm: float
    tasking_imagery_sqkm: Any
    total_imagery_sqkm: float
    estimate: bool

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var estimate : bool
var fresh_imagery_sqkm : float
var standard_imagery_sqkm : float
var tasking_imagery_sqkm : Any
var total_imagery_sqkm : float
var training_imagery_sqkm : float
class UsageAvailable (**data: Any)

A model to hold account balances

When returned with a select, these numbers reflect the account balance at the time the select was submitted. Usage estimates are not applied to these numbers.

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class UsageAvailable(BaseModel):
    """A model to hold account balances

    When returned with a select, these numbers reflect the account balance at the time
    the select was submitted. Usage estimates are not applied to these numbers."""

    # available_sqkm: float # dropped Jan 20
    fresh_imagery_balance: float
    standard_imagery_balance: float
    training_imagery_balance: float
    tasking_imagery_balance: Any
    total_imagery_balance: float

    class Config:
        extra = "allow"

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var Config
var fresh_imagery_balance : float
var standard_imagery_balance : float
var tasking_imagery_balance : Any
var total_imagery_balance : float
var training_imagery_balance : float
class UsageCost (**data: Any)

A model to hold ARD image costs, estimated or consumed

A value of None for a cost means your account does not have pricing set for this category. Without set pricing this category can not be ordered.

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class UsageCost(BaseModel):
    """A model to hold ARD image costs, estimated or consumed

    A value of None for a cost means your account does not have pricing set
    for this category. Without set pricing this category can not be ordered."""

    fresh_imagery_cost: Union[float, None]
    standard_imagery_cost: Union[float, None]
    training_imagery_cost: Union[float, None]
    tasking_imagery_cost: Any
    total_imagery_cost: Union[float, None]
    estimate: bool

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var estimate : bool
var fresh_imagery_cost : Optional[float]
var standard_imagery_cost : Optional[float]
var tasking_imagery_cost : Any
var total_imagery_cost : Optional[float]
var training_imagery_cost : Optional[float]
class UsageLimits (**data: Any)

A model to hold account limits

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code 
class UsageLimits(BaseModel):
    """A model to hold account limits"""

    # limit_sqkm: float # dropped Jan 20
    fresh_imagery_fee_limit: float
    standard_imagery_fee_limit: float
    training_imagery_fee_limit: float
    tasking_imagery_fee_limit: Any
    annual_subscription_fee_limit: float

    class Config:
        extra = "allow"

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var Config
var annual_subscription_fee_limit : float
var fresh_imagery_fee_limit : float
var standard_imagery_fee_limit : float
var tasking_imagery_fee_limit : Any
var training_imagery_fee_limit : float