Module `max_ard.commands.select`

Expand source code

import json
import os
import pprint
import sys
import time
import webbrowser
from collections import defaultdict
from os.path import splitext
from tempfile import NamedTemporaryFile

import click

from max_ard.exceptions import BadARDRequest
from max_ard.io import KmlDoc, KmzDoc, ShpDoc
from max_ard.select import Select


@click.group()
def select():
    """Tools for interacting with the ARD Select service"""
    pass


@click.command()
@click.argument("select_id", required=True)
@click.option(
    "--format",
    help="Return the status in an alternate format. `raw` will return the full response JSON.",
)
def status(select_id, format):
    """Return the status of a Select request using its SELECT_ID"""
    try:
        s = Select.from_id(select_id)
    except Exception as e:
        click.secho(str(e), fg="red", err=True)
        sys.exit()
    if format == "raw":
        click.secho(str(s.response.dict()), fg="cyan")
    else:
        fg_color = "cyan"
        err = False
        if s.state == "SUCCEEDED":
            fg_color = "green"
        elif s.state == "FAILED":
            fg_color = "red"
            err = True
        click.secho(s.state, fg=fg_color, err=err)


@click.command()
@click.argument("select_id", required=True)
@click.option(
    "--dest",
    help="Returns the select in one of the generated format types: `html`, `geojson`, `geojsonl`, `kml`, `kmz`, `shp`, `stac`.",
)
@click.option(
    "--format",
    help="If --dest is not provided, outputs to stdout in these formats:"
    + "`html`, `geojsonl`, `kml`, `stac`. Ignored if --dest is used",
)
@click.option("--verbose", "-v", is_flag=True, help="Return more information about the Select")
def describe(select_id, dest=None, format=None, verbose=False):
    """Returns info about the Select given the SELECT_ID. If no DEST is supplied, prints out a summary of the select.

    Providing a DEST writes to that destination the contents of the given Select result file generated by the system.
    Make sure to include the correct file extension in your desination filepath.
    Formats `geojson`, and `geojsonl`, `kml`, `kmz`, `shp`, and `stac` return files describing the tiles. Format `html` is an interactive map
    browser of the results - to view the map without downloading it first, use `maxard select browse SELECT_ID`"""
    try:
        s = Select.from_id(select_id)
    except Exception as e:
        click.secho(str(e), fg="red", err=True)
        sys.exit()
    pp = pprint.PrettyPrinter(indent=4)
    if not s.finished:
        click.secho("Select has not finished running", fg="cyan")
        if verbose:
            click.secho("*******************", fg="cyan")
            click.secho("Request details:", fg="cyan")
            click.secho(pp.pformat(s.request.to_payload()), fg="cyan")
            click.secho("*******************", fg="cyan")
        exit()
    if s.response.error_message is not None:
        error = s.response.error_message
        click.secho(f'Error in selection process: {error["Error"]}', fg="red", err=True)
        click.secho(f'Reason: {error["Cause"]}', fg="red", err=True)
        exit()
    if dest is None and format is None:
        click.secho("\n")
        click.secho(f"Getting Select ID {select_id}...", fg="cyan")
        click.secho("", fg="cyan")
        click.secho("*******************", fg="cyan")
        click.secho("Request details:", fg="cyan")
        click.secho(pp.pformat(s.request.to_payload()), fg="cyan")
        click.secho("*******************", fg="cyan")
        click.secho("")
        geojson = json.loads(s.get_link_contents("geojson"))
        acqs = defaultdict(list)
        count = 0
        for tile in geojson["features"]:
            for feature in tile["properties"]["best_matches"]:
                count += 1
                acqs[feature["acquisition_id"]].append(feature["date"])
        click.secho(
            f"{count} tiles identified in the following {len(acqs.keys())} acquisitions:",
            fg="green",
        )
        for acq, dates in acqs.items():
            click.secho(f"{acq} ({dates[0]}) - {len(dates)} tiles", fg="green")
        click.secho("")
        click.secho("Ordering this selection will use:", fg="cyan")
        click.secho(f"- {s.usage.area.fresh_imagery_sqkm} sqkm of fresh imagery", fg="cyan")
        click.secho(f"- {s.usage.area.standard_imagery_sqkm} sqkm of standard imagery", fg="cyan")
        click.secho(f"- {s.usage.area.training_imagery_sqkm} sqkm of training imagery", fg="cyan")
        # click.secho('', fg='cyan')
        # click.secho(f'As of {s.usage["usage_as_of"]}:', fg='cyan')
        # click.secho(f'{s.usage["available_sqkm"]} sqkm available', fg='cyan')
        click.secho("")
        exit()
    else:
        if dest:
            format = splitext(dest)[1][1:]

        if format in ["html", "geojson", "geojsonl", "stac", "kml", "kmz", "shp"]:

            # path = Path(dest)

            if (format in s.response.links.keys()) or (
                format == "kml"
            ):  # html, geojson, geojsonl, stac, kml
                if format in s.response.links.keys():
                    file = s.get_link_contents(format)
                else:  # kml
                    file = KmlDoc(s)

                if dest:
                    with open(dest, "w") as out:
                        out.write(file)
                else:
                    click.echo(file)

            if format in ["kmz", "shp"]:
                if dest is None:
                    click.secho(
                        f"Format {format} can not be sent to stdout, use --dest <filename> instead",
                        fg="red",
                        err=True,
                    )
                if format == "kmz":
                    KmzDoc(s, dest)
                else:  # shp
                    ShpDoc(s.get_link_contents("geojson"), dest)

            click.secho(f"Select result file written to {dest}", fg="green")

        else:
            click.secho(
                f'Unknown format {format}, try "html", "geojson", "geojsonl", "kml", "kmz", "shp", or "stac"',
                fg="red",
                err=True,
            )


@click.command()
@click.argument("select_id", required=True)
@click.option(
    "--format",
    help="Format of result file to get signed url for: `html`, `geojson`, geojsonl`, or `stac`",
)
def url(select_id, format):
    """Gets links for the Select result files for a Select of SELECT_ID. If no FORMAT is supplied, returns the base URL of the Select.
    If a FORMAT is supplied, returns a pre-signed URL to download the result of type FORMAT."""
    try:
        s = Select.from_id(select_id)
    except Exception as e:
        click.secho(str(e), fg="red", err=True)
        sys.exit()
    if not s.finished:
        click.secho("Select has not finished running", fg="cyan")
        sys.exit()
    if format is None:
        click.secho(s.response.links["self"], fg="green")
    elif format in s.response.links:
        click.secho(s.get_signed_link(format), fg="green")
    else:
        click.secho(
            f'Unknown format {format}, try "html", "geojson", "geojsonl", or "stac"',
            fg="red",
            err=True,
        )


@click.command()
@click.argument("select_id", required=True)
def browse(select_id):
    """Downloads the HTML interactive viewer of the Select result SELECT ID to a temporary file and opens the
    map with the system web browser.
    When exited the temporary file is deleted. Use `max-ard select describe SELECT_ID --format html > my_map.html`
    to save a local copy"""

    click.secho("Fetching map...", fg="cyan")
    try:
        s = Select.from_id(select_id)
    except Exception as e:
        click.secho(str(e), fg="red", err=True)
        sys.exit()
    if not s.finished:
        click.secho("Select has not finished running", fg="cyan")
        exit()
    prefix = f"ARD_Select_{select_id}-"
    with NamedTemporaryFile(prefix=prefix, suffix=".html", delete=False) as tmp:
        tmp.write(s.get_link_contents("html").encode(encoding="UTF-8"))
    # Windows won't let you open the file from browser while the context manager has it open
    # so we have to manually delete it
    webbrowser.open(f"file://{tmp.name}")
    time.sleep(5)
    os.unlink(tmp.name)


class NumericType(click.ParamType):
    name = "numeric"

    def convert(self, value, param, ctx):
        # strip commas if the got passed in a bbox
        value = value.replace(",", "")
        try:
            return float(value)
        except TypeError:
            self.fail(
                "expected string for int() conversion, got "
                f"{value!r} of type {type(value).__name__}",
                param,
                ctx,
            )
        except ValueError:
            self.fail(f"{value!r} is not a value that can be converted to a float", param, ctx)


NUM_TYPE = NumericType()


@click.command()
@click.option(
    "--acq-id",
    "acq_ids",
    multiple=True,
    help="Limit the Select to these acquisition IDs. Can be provided multiple times",
)
@click.option("--datetime", help="Limit the Select to a given date or date range")
@click.option(
    "--intersects",
    help="Search for tiles that intersect this geometry in WKT format, or load geometry from a file path",
)
@click.option(
    "--bbox",
    nargs=4,
    type=NUM_TYPE,
    help="like `intersects`, but limits search to a WGS84 bounding box in format `--bbox XMIN YMIN XMAX YMAX`",
)
@click.option(
    "--stack-depth",
    type=int,
    help="If provided, only return tiles where the stack depth can be fulfilled.",
)
@click.option(
    "--filter",
    nargs=3,
    multiple=True,
    help="Add a filter statement in the form of `--filter <property> <operator> <value>. See docs for full filter syntax.",
)
@click.option(
    "--image-age",
    "image_age_category",
    multiple=True,
    type=click.Choice(["fresh", "standard", "training"], case_sensitive=False),
    help="Limit imagery to an image age category. Can be used more than once if age ranges are contiguous.",
)
@click.option(
    "--min-cloud-free",
    type=NUM_TYPE,
    help="Shortcut to filter the minimum percentage of cloud-free cover. A value of 100 means an image must be 100% free of clouds.",
)
@click.option(
    "--min-data", type=NUM_TYPE, help="Shortcut to filter the minimum percentage of valid pixels."
)
@click.option(
    "--bba", is_flag=True, help="Restrict imagery to acquisitions that meet BBA requirements"
)
@click.option("--verbose", "-v", is_flag=True, help="Return more information about the Select")
def submit(**kwargs):
    """Submits a Select request to the server.

    There are numerous options and filter possibilities - consult the documentation for more information.

    At a minimum, you must supply at least one of the following:
    - a spatial filter - `intersects` or `bbox`
    - specific acquisition ID or IDs

    The Select service will try to compute results within 20 seconds. If it takes longer than 20 seconds the command
    will return a job ID that you can check for completion with `max-ard select status SELECT_ID`
    """
    # pop out kwargs that don't go the API
    min_cloud_free = kwargs.pop("min_cloud_free")
    min_data = kwargs.pop("min_data")
    filters = kwargs.pop("filter", [])
    for_bba = kwargs.pop("bba", False)
    verbose = kwargs.pop("verbose", False)

    query = defaultdict(dict)
    for prop, op, value in filters:
        # validate props and ops
        if op in ["eq", "ne", "gt", "gte", "lt", "lte"]:
            try:
                value = float(value)
            except:
                pass
        # turn comma-delimited strings to lists
        # and try to cast numbers if needed
        elif op == "in":
            value = value.split(",")
            value = [v.strip() for v in value]
            try:
                value = [float(v) for v in value]
            except:
                pass

        query[prop][op] = value

    # orders with BBA enabled will be rejected if the off-nadir angle of
    # any acquisition is greater than 30 degrees, so don't select images that
    # could cause a failure if the select is ordered
    if for_bba:
        query["view:off_nadir"]["lte"] = 30

    if min_cloud_free is not None:
        if min_cloud_free < 100:
            query["aoi:cloud_free_percentage"]["gte"] = min_cloud_free
        else:
            query["aoi:cloud_free_percentage"]["eq"] = 100

    if min_data is not None:
        if min_data < 100:
            query["aoi:data_percentage"]["gte"] = min_data
        else:
            query["aoi:data_percentage"]["eq"] = 100

    kwargs["query"] = query
    try:
        s = Select(**kwargs)
    except Exception as e:
        click.secho(f"There was an error in your parameters: {e}", fg="red", err=True)
        exit()

    try:
        s.submit()
    except BadARDRequest as e:
        click.secho(f"There was an error in your request: {e}", fg="red", err=True)
        if verbose:
            pp = pprint.PrettyPrinter(indent=4)
            click.secho("Request payload:", fg="red", err=True)
            click.secho(pp.pformat(s.request.to_payload()), fg="red", err=True)
        exit()
    if s.finished:
        click.secho(f"Select {s.select_id} has completed", fg="green")
    else:
        click.secho(f"Select {s.select_id} is still running", fg="cyan")
        click.secho(f"Run `max-ard select status {s.select_id}` to check the status", fg="cyan")
        click.secho('When the status is "SUCCEEDED" you can: ', fg="cyan")
    click.secho(
        f"Run `max-ard select describe {s.select_id}` to see a basic overview of the results",
        fg="cyan",
    )
    click.secho(
        f"Run `max-ard select browse {s.select_id}` to launch a browser-based map view of the results",
        fg="cyan",
    )
    click.secho("Run `max-ard select` to see all the `select` commands", fg="cyan")
    if verbose:
        pp = pprint.PrettyPrinter(indent=4)
        click.secho("Request payload:", fg="cyan")
        click.secho(pp.pformat(s.request.to_payload()), fg="cyan")


select.add_command(status)
select.add_command(describe)
select.add_command(browse)
select.add_command(submit)
select.add_command(url)

Classes

class NumericType

Represents the type of a parameter. Validates and converts values from the command line or Python into the correct type.

To implement a custom type, subclass and implement at least the following:

The :attr:name class attribute must be set.
Calling an instance of the type with None must return None. This is already implemented by default.
:meth:convert must convert string values to the correct type.
:meth:convert must accept values that are already the correct type.
It must be able to convert a value if the ctx and param arguments are None. This can occur when converting prompt input.

Expand source code

class NumericType(click.ParamType):
    name = "numeric"

    def convert(self, value, param, ctx):
        # strip commas if the got passed in a bbox
        value = value.replace(",", "")
        try:
            return float(value)
        except TypeError:
            self.fail(
                "expected string for int() conversion, got "
                f"{value!r} of type {type(value).__name__}",
                param,
                ctx,
            )
        except ValueError:
            self.fail(f"{value!r} is not a value that can be converted to a float", param, ctx)

Ancestors

click.types.ParamType

Class variables

var arity : ClassVar[int]
var envvar_list_splitter : ClassVar[Optional[str]]
var is_composite : ClassVar[bool]
var name : str

Methods

def convert(self, value, param, ctx)

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

:param value: The value to convert. :param param: The parameter that is using this type to convert its value. May be None. :param ctx: The current context that arrived at this value. May be None.

Expand source code

def convert(self, value, param, ctx):
    # strip commas if the got passed in a bbox
    value = value.replace(",", "")
    try:
        return float(value)
    except TypeError:
        self.fail(
            "expected string for int() conversion, got "
            f"{value!r} of type {type(value).__name__}",
            param,
            ctx,
        )
    except ValueError:
        self.fail(f"{value!r} is not a value that can be converted to a float", param, ctx)