appeears module¶

Utilities for submitting and downloading NASA AppEEARS tasks.

`AppEEARSClient` ¶

Client for the NASA AppEEARS API.

Parameters:

Name	Type	Description	Default
`api_url`	`str`	Base URL for the AppEEARS API.	`'https://appeears.earthdatacloud.nasa.gov/api/'`
`token`	`Optional[str]`	Existing AppEEARS bearer token.	`None`
`session`	`Optional[Any]`	Optional requests-compatible session.	`None`

Source code in hypercoast/appeears.py

class AppEEARSClient:
    """Client for the NASA AppEEARS API.

    Args:
        api_url: Base URL for the AppEEARS API.
        token: Existing AppEEARS bearer token.
        session: Optional requests-compatible session.
    """

    def __init__(
        self,
        api_url: str = APPEEARS_API_URL,
        token: Optional[str] = None,
        session: Optional[Any] = None,
    ) -> None:
        import requests

        self.api_url = api_url.rstrip("/") + "/"
        self.token = token
        self.session = session or requests.Session()

    @property
    def headers(self) -> Dict[str, str]:
        """Return authorization headers for authenticated requests.

        Returns:
            Dictionary containing the bearer token header when a token exists.
        """

        if self.token is None:
            return {}
        return {"Authorization": f"Bearer {self.token}"}

    def login(
        self,
        username: Optional[str] = None,
        password: Optional[str] = None,
        token: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Authenticate with AppEEARS.

        Args:
            username: NASA Earthdata username. If omitted, environment variables
                or a netrc entry for urs.earthdata.nasa.gov are used.
            password: NASA Earthdata password.
            token: Existing AppEEARS token. When provided, no login request is
                sent.

        Returns:
            AppEEARS login response containing the bearer token.
        """

        if token is not None:
            self.token = token
            return {"token": token}

        if username is None:
            username = os.environ.get("EARTHDATA_USERNAME")
        if password is None:
            password = os.environ.get("EARTHDATA_PASSWORD")

        if username is None or password is None:
            try:
                from netrc import netrc

                auth = netrc().authenticators("urs.earthdata.nasa.gov")
            except (FileNotFoundError, OSError):
                auth = None
            if auth is not None:
                username = username or auth[0]
                password = password or auth[2]

        if username is None or password is None:
            raise ValueError(
                "username and password are required unless EARTHDATA_USERNAME "
                "and EARTHDATA_PASSWORD or a netrc entry are available."
            )

        response = self._request(
            "post", "login", auth=(username, password), auth_header=False
        )
        self.token = response["token"]
        return response

    def products(
        self,
        keyword: Optional[str] = None,
        platform: Optional[str] = None,
        available: Optional[bool] = True,
    ) -> List[Dict[str, Any]]:
        """List AppEEARS products.

        Args:
            keyword: Case-insensitive search text matched against product fields.
            platform: Platform name, such as ``"EMIT"``.
            available: If set, filter by the AppEEARS availability flag.

        Returns:
            List of product metadata dictionaries.
        """

        products = self._request("get", "product", auth_header=False)
        if keyword is not None:
            query = keyword.lower()
            products = [p for p in products if query in json.dumps(p).lower()]
        if platform is not None:
            products = [
                p
                for p in products
                if str(p.get("Platform", "")).lower() == platform.lower()
            ]
        if available is not None:
            products = [p for p in products if p.get("Available") is available]
        return products

    def layers(self, product: str) -> Dict[str, Dict[str, Any]]:
        """List layers for an AppEEARS product.

        Args:
            product: Product and version string, such as ``"EMIT_L2A_RFL.001"``.

        Returns:
            Dictionary keyed by layer name.
        """

        return self._request("get", f"product/{product}", auth_header=False)

    def spectral_layers(
        self,
        product: str = EMIT_REFLECTANCE_PRODUCT,
        wavelengths: Optional[Sequence[float]] = None,
        include_flags: bool = False,
    ) -> List[Dict[str, Any]]:
        """Return spectral layer metadata, optionally nearest to wavelengths.

        Args:
            product: AppEEARS product and version.
            wavelengths: Target wavelengths in nanometers. If omitted, all
                spectral bands are returned.
            include_flags: Whether to include good wavelength flag layers.

        Returns:
            List of layer records containing product, layer, wavelength, fwhm,
            units, description, and data type.
        """

        records = []
        for layer, metadata in self.layers(product).items():
            if not include_flags and layer.endswith("_GW"):
                continue
            pattern = r"B\d{3}(?:_GW)?" if include_flags else r"B\d{3}"
            if not re.fullmatch(pattern, layer):
                continue
            record = _layer_record(product, layer, metadata)
            if record["wavelength"] is not None:
                records.append(record)

        records.sort(key=lambda item: item["wavelength"])
        if wavelengths is None:
            return records

        selected = []
        used_layers = set()
        for wavelength in wavelengths:
            nearest = min(
                records,
                key=lambda item: abs(float(item["wavelength"]) - float(wavelength)),
            )
            if nearest["layer"] not in used_layers:
                selected.append(nearest)
                used_layers.add(nearest["layer"])
        return selected

    def projections(self) -> Dict[str, Any]:
        """List AppEEARS output projections.

        Returns:
            Dictionary of AppEEARS projection metadata.
        """

        return self._request("get", "spatial/proj", auth_header=False)

    def submit_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """Submit an AppEEARS task.

        Args:
            task: AppEEARS task payload.

        Returns:
            AppEEARS task submission response.
        """

        self._require_token()
        return self._request("post", "task", json=task)

    def tasks(
        self,
        limit: Optional[int] = None,
        offset: Optional[int] = None,
        pretty: bool = False,
    ) -> Dict[str, Any]:
        """List submitted AppEEARS tasks for the authenticated user.

        Args:
            limit: Maximum number of tasks to return.
            offset: Result offset for pagination.
            pretty: Whether AppEEARS should pretty-print the JSON response.

        Returns:
            Task list response from AppEEARS.
        """

        self._require_token()
        params = {"pretty": pretty}
        if limit is not None:
            params["limit"] = limit
        if offset is not None:
            params["offset"] = offset
        return self._request("get", "task", params=params)

    def task(self, task_id: str) -> Dict[str, Any]:
        """Get an AppEEARS task.

        Args:
            task_id: AppEEARS task identifier.

        Returns:
            Task metadata response.
        """

        self._require_token()
        return self._request("get", f"task/{task_id}")

    def status(self, task_id: str) -> Dict[str, Any]:
        """Get an AppEEARS task status.

        Args:
            task_id: AppEEARS task identifier.

        Returns:
            Task status response.
        """

        self._require_token()
        return self._request("get", f"status/{task_id}")

    def wait_for_task(
        self,
        task_id: str,
        interval: int = 20,
        timeout: Optional[int] = None,
        verbose: bool = True,
    ) -> Dict[str, Any]:
        """Wait for an AppEEARS task to finish.

        Args:
            task_id: AppEEARS task identifier.
            interval: Polling interval in seconds.
            timeout: Optional timeout in seconds.
            verbose: Whether to print status changes.

        Returns:
            Final task metadata response.
        """

        start = time.time()
        last_status = None
        while True:
            response = self.task(task_id)
            status = response.get("status")
            if verbose and status != last_status:
                print(status)
            if status == "done":
                return response
            if status in {"error", "failed"}:
                raise RuntimeError(f"AppEEARS task {task_id} failed: {response}")
            if timeout is not None and time.time() - start > timeout:
                raise TimeoutError(f"Timed out waiting for AppEEARS task {task_id}.")
            last_status = status
            time.sleep(interval)

    def bundle(self, task_id: str) -> Dict[str, Any]:
        """Get bundle metadata for a completed AppEEARS task.

        Args:
            task_id: AppEEARS task identifier.

        Returns:
            Bundle metadata response.
        """

        self._require_token()
        return self._request("get", f"bundle/{task_id}")

    def download_bundle(
        self,
        task_id: str,
        out_dir: Union[str, os.PathLike[str]] = ".",
        file_ids: Optional[Iterable[str]] = None,
        file_types: Optional[Iterable[str]] = None,
        overwrite: bool = False,
    ) -> List[str]:
        """Download files from an AppEEARS task bundle.

        Args:
            task_id: AppEEARS task identifier.
            out_dir: Output directory.
            file_ids: Optional subset of AppEEARS file identifiers to download.
            file_types: Optional file type filter, such as ``["tif", "csv"]``.
            overwrite: Whether to overwrite existing files.

        Returns:
            List of downloaded file paths.
        """

        bundle = self.bundle(task_id)
        selected_file_ids = set(file_ids) if file_ids is not None else None
        selected_file_types = set(file_types) if file_types is not None else None
        paths = []
        for item in bundle.get("files", []):
            file_id = item["file_id"]
            if selected_file_ids is not None and file_id not in selected_file_ids:
                continue
            if (
                selected_file_types is not None
                and item.get("file_type") not in selected_file_types
            ):
                continue
            path = self.download_file(
                task_id=task_id,
                file_id=file_id,
                file_name=item["file_name"],
                out_dir=out_dir,
                overwrite=overwrite,
            )
            paths.append(path)
        return paths

    def download_file(
        self,
        task_id: str,
        file_id: str,
        file_name: str,
        out_dir: Union[str, os.PathLike[str]] = ".",
        overwrite: bool = False,
        chunk_size: int = 8192,
    ) -> str:
        """Download one AppEEARS bundle file.

        Args:
            task_id: AppEEARS task identifier.
            file_id: Bundle file identifier.
            file_name: Bundle file name.
            out_dir: Output directory.
            overwrite: Whether to overwrite an existing file.
            chunk_size: Response chunk size in bytes.

        Returns:
            Path to the downloaded file.
        """

        self._require_token()
        out_path = _bundle_output_path(out_dir, file_name)
        if out_path.exists() and not overwrite:
            return str(out_path)

        response = self._raw_request("get", f"bundle/{task_id}/{file_id}", stream=True)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        with out_path.open("wb") as dst:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if chunk:
                    dst.write(chunk)
        return str(out_path)

    def s3credentials(self) -> Dict[str, Any]:
        """Request temporary S3 credentials from AppEEARS.

        Returns:
            Temporary credential response from AppEEARS.
        """

        self._require_token()
        return self._request("get", "s3credentials")

    def _request(
        self,
        method: str,
        path: str,
        auth_header: bool = True,
        **kwargs: Any,
    ) -> Any:
        """Send a JSON request to AppEEARS.

        Args:
            method: HTTP method name.
            path: API path relative to the base URL.
            auth_header: Whether to include the bearer token header.
            **kwargs: Additional request keyword arguments.

        Returns:
            Decoded JSON response.
        """

        response = self._raw_request(method, path, auth_header=auth_header, **kwargs)
        try:
            return response.json()
        except ValueError:
            return response.text

    def _raw_request(
        self,
        method: str,
        path: str,
        auth_header: bool = True,
        **kwargs: Any,
    ) -> Any:
        """Send a raw request to AppEEARS.

        Args:
            method: HTTP method name.
            path: API path relative to the base URL.
            auth_header: Whether to include the bearer token header.
            **kwargs: Additional request keyword arguments.

        Returns:
            Requests response object.
        """

        headers = kwargs.pop("headers", {})
        if auth_header:
            headers = {**self.headers, **headers}
        kwargs.setdefault("timeout", 120)
        response = self.session.request(
            method.upper(), self.api_url + path.lstrip("/"), headers=headers, **kwargs
        )
        try:
            response.raise_for_status()
        except Exception as error:
            message = _response_error_message(response)
            if message:
                error.args = (*error.args, message)
            raise
        return response

    def _require_token(self) -> None:
        """Raise an error when an authenticated request lacks a token."""

        if self.token is None:
            raise ValueError("Call login() or provide a token before this request.")

`headers: Dict[str, str]` `property` `readonly` ¶

Return authorization headers for authenticated requests.

Returns:

Type	Description
`Dict[str, str]`	Dictionary containing the bearer token header when a token exists.

`bundle(self, task_id)` ¶

Get bundle metadata for a completed AppEEARS task.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required

Returns:

Type	Description
`Dict[str, Any]`	Bundle metadata response.

Source code in hypercoast/appeears.py

def bundle(self, task_id: str) -> Dict[str, Any]:
    """Get bundle metadata for a completed AppEEARS task.

    Args:
        task_id: AppEEARS task identifier.

    Returns:
        Bundle metadata response.
    """

    self._require_token()
    return self._request("get", f"bundle/{task_id}")

`download_bundle(self, task_id, out_dir='.', file_ids=None, file_types=None, overwrite=False)` ¶

Download files from an AppEEARS task bundle.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required
`out_dir`	`Union[str, os.PathLike[str]]`	Output directory.	`'.'`
`file_ids`	`Optional[Iterable[str]]`	Optional subset of AppEEARS file identifiers to download.	`None`
`file_types`	`Optional[Iterable[str]]`	Optional file type filter, such as `["tif", "csv"]`.	`None`
`overwrite`	`bool`	Whether to overwrite existing files.	`False`

Returns:

Type	Description
`List[str]`	List of downloaded file paths.

Source code in hypercoast/appeears.py

def download_bundle(
    self,
    task_id: str,
    out_dir: Union[str, os.PathLike[str]] = ".",
    file_ids: Optional[Iterable[str]] = None,
    file_types: Optional[Iterable[str]] = None,
    overwrite: bool = False,
) -> List[str]:
    """Download files from an AppEEARS task bundle.

    Args:
        task_id: AppEEARS task identifier.
        out_dir: Output directory.
        file_ids: Optional subset of AppEEARS file identifiers to download.
        file_types: Optional file type filter, such as ``["tif", "csv"]``.
        overwrite: Whether to overwrite existing files.

    Returns:
        List of downloaded file paths.
    """

    bundle = self.bundle(task_id)
    selected_file_ids = set(file_ids) if file_ids is not None else None
    selected_file_types = set(file_types) if file_types is not None else None
    paths = []
    for item in bundle.get("files", []):
        file_id = item["file_id"]
        if selected_file_ids is not None and file_id not in selected_file_ids:
            continue
        if (
            selected_file_types is not None
            and item.get("file_type") not in selected_file_types
        ):
            continue
        path = self.download_file(
            task_id=task_id,
            file_id=file_id,
            file_name=item["file_name"],
            out_dir=out_dir,
            overwrite=overwrite,
        )
        paths.append(path)
    return paths

`download_file(self, task_id, file_id, file_name, out_dir='.', overwrite=False, chunk_size=8192)` ¶

Download one AppEEARS bundle file.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required
`file_id`	`str`	Bundle file identifier.	required
`file_name`	`str`	Bundle file name.	required
`out_dir`	`Union[str, os.PathLike[str]]`	Output directory.	`'.'`
`overwrite`	`bool`	Whether to overwrite an existing file.	`False`
`chunk_size`	`int`	Response chunk size in bytes.	`8192`

Returns:

Type	Description
`str`	Path to the downloaded file.

Source code in hypercoast/appeears.py

def download_file(
    self,
    task_id: str,
    file_id: str,
    file_name: str,
    out_dir: Union[str, os.PathLike[str]] = ".",
    overwrite: bool = False,
    chunk_size: int = 8192,
) -> str:
    """Download one AppEEARS bundle file.

    Args:
        task_id: AppEEARS task identifier.
        file_id: Bundle file identifier.
        file_name: Bundle file name.
        out_dir: Output directory.
        overwrite: Whether to overwrite an existing file.
        chunk_size: Response chunk size in bytes.

    Returns:
        Path to the downloaded file.
    """

    self._require_token()
    out_path = _bundle_output_path(out_dir, file_name)
    if out_path.exists() and not overwrite:
        return str(out_path)

    response = self._raw_request("get", f"bundle/{task_id}/{file_id}", stream=True)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("wb") as dst:
        for chunk in response.iter_content(chunk_size=chunk_size):
            if chunk:
                dst.write(chunk)
    return str(out_path)

`layers(self, product)` ¶

List layers for an AppEEARS product.

Parameters:

Name	Type	Description	Default
`product`	`str`	Product and version string, such as `"EMIT_L2A_RFL.001"`.	required

Returns:

Type	Description
`Dict[str, Dict[str, Any]]`	Dictionary keyed by layer name.

Source code in hypercoast/appeears.py

def layers(self, product: str) -> Dict[str, Dict[str, Any]]:
    """List layers for an AppEEARS product.

    Args:
        product: Product and version string, such as ``"EMIT_L2A_RFL.001"``.

    Returns:
        Dictionary keyed by layer name.
    """

    return self._request("get", f"product/{product}", auth_header=False)

`login(self, username=None, password=None, token=None)` ¶

Authenticate with AppEEARS.

Parameters:

Name	Type	Description	Default
`username`	`Optional[str]`	NASA Earthdata username. If omitted, environment variables or a netrc entry for urs.earthdata.nasa.gov are used.	`None`
`password`	`Optional[str]`	NASA Earthdata password.	`None`
`token`	`Optional[str]`	Existing AppEEARS token. When provided, no login request is sent.	`None`

Returns:

Type	Description
`Dict[str, Any]`	AppEEARS login response containing the bearer token.

Source code in hypercoast/appeears.py

def login(
    self,
    username: Optional[str] = None,
    password: Optional[str] = None,
    token: Optional[str] = None,
) -> Dict[str, Any]:
    """Authenticate with AppEEARS.

    Args:
        username: NASA Earthdata username. If omitted, environment variables
            or a netrc entry for urs.earthdata.nasa.gov are used.
        password: NASA Earthdata password.
        token: Existing AppEEARS token. When provided, no login request is
            sent.

    Returns:
        AppEEARS login response containing the bearer token.
    """

    if token is not None:
        self.token = token
        return {"token": token}

    if username is None:
        username = os.environ.get("EARTHDATA_USERNAME")
    if password is None:
        password = os.environ.get("EARTHDATA_PASSWORD")

    if username is None or password is None:
        try:
            from netrc import netrc

            auth = netrc().authenticators("urs.earthdata.nasa.gov")
        except (FileNotFoundError, OSError):
            auth = None
        if auth is not None:
            username = username or auth[0]
            password = password or auth[2]

    if username is None or password is None:
        raise ValueError(
            "username and password are required unless EARTHDATA_USERNAME "
            "and EARTHDATA_PASSWORD or a netrc entry are available."
        )

    response = self._request(
        "post", "login", auth=(username, password), auth_header=False
    )
    self.token = response["token"]
    return response

`products(self, keyword=None, platform=None, available=True)` ¶

List AppEEARS products.

Parameters:

Name	Type	Description	Default
`keyword`	`Optional[str]`	Case-insensitive search text matched against product fields.	`None`
`platform`	`Optional[str]`	Platform name, such as `"EMIT"`.	`None`
`available`	`Optional[bool]`	If set, filter by the AppEEARS availability flag.	`True`

Returns:

Type	Description
`List[Dict[str, Any]]`	List of product metadata dictionaries.

Source code in hypercoast/appeears.py

def products(
    self,
    keyword: Optional[str] = None,
    platform: Optional[str] = None,
    available: Optional[bool] = True,
) -> List[Dict[str, Any]]:
    """List AppEEARS products.

    Args:
        keyword: Case-insensitive search text matched against product fields.
        platform: Platform name, such as ``"EMIT"``.
        available: If set, filter by the AppEEARS availability flag.

    Returns:
        List of product metadata dictionaries.
    """

    products = self._request("get", "product", auth_header=False)
    if keyword is not None:
        query = keyword.lower()
        products = [p for p in products if query in json.dumps(p).lower()]
    if platform is not None:
        products = [
            p
            for p in products
            if str(p.get("Platform", "")).lower() == platform.lower()
        ]
    if available is not None:
        products = [p for p in products if p.get("Available") is available]
    return products

`projections(self)` ¶

List AppEEARS output projections.

Returns:

Type	Description
`Dict[str, Any]`	Dictionary of AppEEARS projection metadata.

Source code in hypercoast/appeears.py

def projections(self) -> Dict[str, Any]:
    """List AppEEARS output projections.

    Returns:
        Dictionary of AppEEARS projection metadata.
    """

    return self._request("get", "spatial/proj", auth_header=False)

`s3credentials(self)` ¶

Request temporary S3 credentials from AppEEARS.

Returns:

Type	Description
`Dict[str, Any]`	Temporary credential response from AppEEARS.

Source code in hypercoast/appeears.py

def s3credentials(self) -> Dict[str, Any]:
    """Request temporary S3 credentials from AppEEARS.

    Returns:
        Temporary credential response from AppEEARS.
    """

    self._require_token()
    return self._request("get", "s3credentials")

`spectral_layers(self, product='EMIT_L2A_RFL.001', wavelengths=None, include_flags=False)` ¶

Return spectral layer metadata, optionally nearest to wavelengths.

Parameters:

Name	Type	Description	Default
`product`	`str`	AppEEARS product and version.	`'EMIT_L2A_RFL.001'`
`wavelengths`	`Optional[Sequence[float]]`	Target wavelengths in nanometers. If omitted, all spectral bands are returned.	`None`
`include_flags`	`bool`	Whether to include good wavelength flag layers.	`False`

Returns:

Type	Description
`List[Dict[str, Any]]`	List of layer records containing product, layer, wavelength, fwhm, units, description, and data type.

Source code in hypercoast/appeears.py

def spectral_layers(
    self,
    product: str = EMIT_REFLECTANCE_PRODUCT,
    wavelengths: Optional[Sequence[float]] = None,
    include_flags: bool = False,
) -> List[Dict[str, Any]]:
    """Return spectral layer metadata, optionally nearest to wavelengths.

    Args:
        product: AppEEARS product and version.
        wavelengths: Target wavelengths in nanometers. If omitted, all
            spectral bands are returned.
        include_flags: Whether to include good wavelength flag layers.

    Returns:
        List of layer records containing product, layer, wavelength, fwhm,
        units, description, and data type.
    """

    records = []
    for layer, metadata in self.layers(product).items():
        if not include_flags and layer.endswith("_GW"):
            continue
        pattern = r"B\d{3}(?:_GW)?" if include_flags else r"B\d{3}"
        if not re.fullmatch(pattern, layer):
            continue
        record = _layer_record(product, layer, metadata)
        if record["wavelength"] is not None:
            records.append(record)

    records.sort(key=lambda item: item["wavelength"])
    if wavelengths is None:
        return records

    selected = []
    used_layers = set()
    for wavelength in wavelengths:
        nearest = min(
            records,
            key=lambda item: abs(float(item["wavelength"]) - float(wavelength)),
        )
        if nearest["layer"] not in used_layers:
            selected.append(nearest)
            used_layers.add(nearest["layer"])
    return selected

`status(self, task_id)` ¶

Get an AppEEARS task status.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required

Returns:

Type	Description
`Dict[str, Any]`	Task status response.

Source code in hypercoast/appeears.py

def status(self, task_id: str) -> Dict[str, Any]:
    """Get an AppEEARS task status.

    Args:
        task_id: AppEEARS task identifier.

    Returns:
        Task status response.
    """

    self._require_token()
    return self._request("get", f"status/{task_id}")

`submit_task(self, task)` ¶

Submit an AppEEARS task.

Parameters:

Name	Type	Description	Default
`task`	`Dict[str, Any]`	AppEEARS task payload.	required

Returns:

Type	Description
`Dict[str, Any]`	AppEEARS task submission response.

Source code in hypercoast/appeears.py

def submit_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
    """Submit an AppEEARS task.

    Args:
        task: AppEEARS task payload.

    Returns:
        AppEEARS task submission response.
    """

    self._require_token()
    return self._request("post", "task", json=task)

`task(self, task_id)` ¶

Get an AppEEARS task.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required

Returns:

Type	Description
`Dict[str, Any]`	Task metadata response.

Source code in hypercoast/appeears.py

def task(self, task_id: str) -> Dict[str, Any]:
    """Get an AppEEARS task.

    Args:
        task_id: AppEEARS task identifier.

    Returns:
        Task metadata response.
    """

    self._require_token()
    return self._request("get", f"task/{task_id}")

`tasks(self, limit=None, offset=None, pretty=False)` ¶

List submitted AppEEARS tasks for the authenticated user.

Parameters:

Name	Type	Description	Default
`limit`	`Optional[int]`	Maximum number of tasks to return.	`None`
`offset`	`Optional[int]`	Result offset for pagination.	`None`
`pretty`	`bool`	Whether AppEEARS should pretty-print the JSON response.	`False`

Returns:

Type	Description
`Dict[str, Any]`	Task list response from AppEEARS.

Source code in hypercoast/appeears.py

def tasks(
    self,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
    pretty: bool = False,
) -> Dict[str, Any]:
    """List submitted AppEEARS tasks for the authenticated user.

    Args:
        limit: Maximum number of tasks to return.
        offset: Result offset for pagination.
        pretty: Whether AppEEARS should pretty-print the JSON response.

    Returns:
        Task list response from AppEEARS.
    """

    self._require_token()
    params = {"pretty": pretty}
    if limit is not None:
        params["limit"] = limit
    if offset is not None:
        params["offset"] = offset
    return self._request("get", "task", params=params)

`wait_for_task(self, task_id, interval=20, timeout=None, verbose=True)` ¶

Wait for an AppEEARS task to finish.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required
`interval`	`int`	Polling interval in seconds.	`20`
`timeout`	`Optional[int]`	Optional timeout in seconds.	`None`
`verbose`	`bool`	Whether to print status changes.	`True`

Returns:

Type	Description
`Dict[str, Any]`	Final task metadata response.

Source code in hypercoast/appeears.py

def wait_for_task(
    self,
    task_id: str,
    interval: int = 20,
    timeout: Optional[int] = None,
    verbose: bool = True,
) -> Dict[str, Any]:
    """Wait for an AppEEARS task to finish.

    Args:
        task_id: AppEEARS task identifier.
        interval: Polling interval in seconds.
        timeout: Optional timeout in seconds.
        verbose: Whether to print status changes.

    Returns:
        Final task metadata response.
    """

    start = time.time()
    last_status = None
    while True:
        response = self.task(task_id)
        status = response.get("status")
        if verbose and status != last_status:
            print(status)
        if status == "done":
            return response
        if status in {"error", "failed"}:
            raise RuntimeError(f"AppEEARS task {task_id} failed: {response}")
        if timeout is not None and time.time() - start > timeout:
            raise TimeoutError(f"Timed out waiting for AppEEARS task {task_id}.")
        last_status = status
        time.sleep(interval)

`appeears_area_task(task_name, geometry, layers, start_date, end_date, output_format=None, projection='geographic', recurring=False, year_range=None, orthorectify=True, additional_options=None)` ¶

Build an AppEEARS area task payload.

Parameters:

Name	Type	Description	Default
`task_name`	`str`	AppEEARS task name.	required
`geometry`	`Any`	GeoJSON geometry, feature, feature collection, GeoDataFrame, vector file path, or bounding box `[xmin, ymin, xmax, ymax]`.	required
`layers`	`Sequence[Dict[str, str]]`	Task-ready AppEEARS layers.	required
`start_date`	`str`	Start date as `YYYY-MM-DD` or `MM-DD-YYYY`.	required
`end_date`	`str`	End date as `YYYY-MM-DD` or `MM-DD-YYYY`.	required
`output_format`	`Optional[str]`	AppEEARS output format, such as `"geotiff"` or `"netcdf4"`. If None, EMIT tasks use `"netcdf4"` and other tasks use `"geotiff"`.	`None`
`projection`	`str`	AppEEARS output projection name.	`'geographic'`
`recurring`	`bool`	Whether the task uses recurring dates.	`False`
`year_range`	`Optional[Sequence[int]]`	Two-year range for recurring tasks.	`None`
`orthorectify`	`Optional[bool]`	Whether AppEEARS should orthorectify EMIT area outputs. AppEEARS requires this option for EMIT area tasks. Set to None to omit it.	`True`
`additional_options`	`Optional[Dict[str, Any]]`	Additional AppEEARS output options.	`None`

Returns:

Type	Description
`Dict[str, Any]`	AppEEARS area task payload.

Source code in hypercoast/appeears.py

def appeears_area_task(
    task_name: str,
    geometry: Any,
    layers: Sequence[Dict[str, str]],
    start_date: str,
    end_date: str,
    output_format: Optional[str] = None,
    projection: str = "geographic",
    recurring: bool = False,
    year_range: Optional[Sequence[int]] = None,
    orthorectify: Optional[bool] = True,
    additional_options: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Build an AppEEARS area task payload.

    Args:
        task_name: AppEEARS task name.
        geometry: GeoJSON geometry, feature, feature collection, GeoDataFrame,
            vector file path, or bounding box ``[xmin, ymin, xmax, ymax]``.
        layers: Task-ready AppEEARS layers.
        start_date: Start date as ``YYYY-MM-DD`` or ``MM-DD-YYYY``.
        end_date: End date as ``YYYY-MM-DD`` or ``MM-DD-YYYY``.
        output_format: AppEEARS output format, such as ``"geotiff"`` or
            ``"netcdf4"``. If None, EMIT tasks use ``"netcdf4"`` and other
            tasks use ``"geotiff"``.
        projection: AppEEARS output projection name.
        recurring: Whether the task uses recurring dates.
        year_range: Two-year range for recurring tasks.
        orthorectify: Whether AppEEARS should orthorectify EMIT area outputs.
            AppEEARS requires this option for EMIT area tasks. Set to None to
            omit it.
        additional_options: Additional AppEEARS output options.

    Returns:
        AppEEARS area task payload.
    """

    has_emit = _has_emit_layers(layers)
    if output_format is None:
        output_format = "netcdf4" if has_emit else "geotiff"

    output = {"format": {"type": output_format}, "projection": projection}
    options = dict(additional_options or {})
    if orthorectify is not None and has_emit:
        options.setdefault("orthorectify", bool(orthorectify))
    if options:
        output["additionalOptions"] = options

    params = {
        "dates": [
            _date_range(
                start_date=start_date,
                end_date=end_date,
                recurring=recurring,
                year_range=year_range,
            )
        ],
        "layers": list(layers),
        "output": output,
        "geo": _normalize_geojson(geometry),
    }
    return {"task_type": "area", "task_name": task_name, "params": params}

`appeears_download(task_id, out_dir='.', client=None, file_types=None, overwrite=False)` ¶

Download an AppEEARS task bundle.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required
`out_dir`	`Union[str, os.PathLike[str]]`	Output directory.	`'.'`
`client`	`Optional[AppEEARSClient]`	Authenticated AppEEARS client.	`None`
`file_types`	`Optional[Iterable[str]]`	Optional file type filter, such as `["tif"]`.	`None`
`overwrite`	`bool`	Whether to overwrite existing files.	`False`

Returns:

Type	Description
`List[str]`	List of downloaded file paths.

Source code in hypercoast/appeears.py

def appeears_download(
    task_id: str,
    out_dir: Union[str, os.PathLike[str]] = ".",
    client: Optional[AppEEARSClient] = None,
    file_types: Optional[Iterable[str]] = None,
    overwrite: bool = False,
) -> List[str]:
    """Download an AppEEARS task bundle.

    Args:
        task_id: AppEEARS task identifier.
        out_dir: Output directory.
        client: Authenticated AppEEARS client.
        file_types: Optional file type filter, such as ``["tif"]``.
        overwrite: Whether to overwrite existing files.

    Returns:
        List of downloaded file paths.
    """

    if client is None:
        client = appeears_login()
    return client.download_bundle(
        task_id=task_id,
        out_dir=out_dir,
        file_types=file_types,
        overwrite=overwrite,
    )

`appeears_emit_layers(wavelengths=None, product='EMIT_L2A_RFL.001', include_flags=False, return_metadata=False, client=None)` ¶

Select EMIT AppEEARS layers by wavelength.

Parameters:

Name	Type	Description	Default
`wavelengths`	`Optional[Sequence[float]]`	Target wavelengths in nanometers. If omitted, all spectral EMIT bands are returned.	`None`
`product`	`str`	EMIT product and version.	`'EMIT_L2A_RFL.001'`
`include_flags`	`bool`	Whether to include good wavelength flag layers.	`False`
`return_metadata`	`bool`	Whether to return wavelength metadata. If False, task-ready `{"product": product, "layer": layer}` dictionaries are returned.	`False`
`client`	`Optional[AppEEARSClient]`	Optional AppEEARS client.	`None`

Returns:

Type	Description
`List[Dict[str, Any]]`	Selected EMIT layer records or task-ready layer dictionaries.

Source code in hypercoast/appeears.py

def appeears_emit_layers(
    wavelengths: Optional[Sequence[float]] = None,
    product: str = EMIT_REFLECTANCE_PRODUCT,
    include_flags: bool = False,
    return_metadata: bool = False,
    client: Optional[AppEEARSClient] = None,
) -> List[Dict[str, Any]]:
    """Select EMIT AppEEARS layers by wavelength.

    Args:
        wavelengths: Target wavelengths in nanometers. If omitted, all spectral
            EMIT bands are returned.
        product: EMIT product and version.
        include_flags: Whether to include good wavelength flag layers.
        return_metadata: Whether to return wavelength metadata. If False,
            task-ready ``{"product": product, "layer": layer}`` dictionaries
            are returned.
        client: Optional AppEEARS client.

    Returns:
        Selected EMIT layer records or task-ready layer dictionaries.
    """

    client = client or AppEEARSClient()
    layers = client.spectral_layers(
        product=product, wavelengths=wavelengths, include_flags=include_flags
    )
    if return_metadata:
        return layers
    return [{"product": item["product"], "layer": item["layer"]} for item in layers]

`appeears_layers(product, client=None)` ¶

List AppEEARS layers for a product.

Parameters:

Name	Type	Description	Default
`product`	`str`	Product and version string.	required
`client`	`Optional[AppEEARSClient]`	Optional AppEEARS client.	`None`

Returns:

Type	Description
`Dict[str, Dict[str, Any]]`	Dictionary keyed by layer name.

Source code in hypercoast/appeears.py

def appeears_layers(
    product: str,
    client: Optional[AppEEARSClient] = None,
) -> Dict[str, Dict[str, Any]]:
    """List AppEEARS layers for a product.

    Args:
        product: Product and version string.
        client: Optional AppEEARS client.

    Returns:
        Dictionary keyed by layer name.
    """

    client = client or AppEEARSClient()
    return client.layers(product)

`appeears_login(username=None, password=None, token=None, api_url='https://appeears.earthdatacloud.nasa.gov/api/')` ¶

Create and authenticate an AppEEARS client.

Parameters:

Name	Type	Description	Default
`username`	`Optional[str]`	NASA Earthdata username.	`None`
`password`	`Optional[str]`	NASA Earthdata password.	`None`
`token`	`Optional[str]`	Existing AppEEARS token.	`None`
`api_url`	`str`	Base URL for the AppEEARS API.	`'https://appeears.earthdatacloud.nasa.gov/api/'`

Returns:

Type	Description
`AppEEARSClient`	Authenticated AppEEARS client.

Source code in hypercoast/appeears.py

def appeears_login(
    username: Optional[str] = None,
    password: Optional[str] = None,
    token: Optional[str] = None,
    api_url: str = APPEEARS_API_URL,
) -> AppEEARSClient:
    """Create and authenticate an AppEEARS client.

    Args:
        username: NASA Earthdata username.
        password: NASA Earthdata password.
        token: Existing AppEEARS token.
        api_url: Base URL for the AppEEARS API.

    Returns:
        Authenticated AppEEARS client.
    """

    client = AppEEARSClient(api_url=api_url)
    client.login(username=username, password=password, token=token)
    return client

`appeears_point_task(task_name, coordinates, layers, start_date, end_date, recurring=False, year_range=None)` ¶

Build an AppEEARS point task payload.

Parameters:

Name	Type	Description	Default
`task_name`	`str`	AppEEARS task name.	required
`coordinates`	`Union[Sequence[float], Sequence[Dict[str, Any]]]`	One `(longitude, latitude)` pair or AppEEARS coordinate dictionaries containing longitude and latitude.	required
`layers`	`Sequence[Dict[str, str]]`	Task-ready AppEEARS layers.	required
`start_date`	`str`	Start date as `YYYY-MM-DD` or `MM-DD-YYYY`.	required
`end_date`	`str`	End date as `YYYY-MM-DD` or `MM-DD-YYYY`.	required
`recurring`	`bool`	Whether the task uses recurring dates.	`False`
`year_range`	`Optional[Sequence[int]]`	Two-year range for recurring tasks.	`None`

Returns:

Type	Description
`Dict[str, Any]`	AppEEARS point task payload.

Source code in hypercoast/appeears.py

def appeears_point_task(
    task_name: str,
    coordinates: Union[Sequence[float], Sequence[Dict[str, Any]]],
    layers: Sequence[Dict[str, str]],
    start_date: str,
    end_date: str,
    recurring: bool = False,
    year_range: Optional[Sequence[int]] = None,
) -> Dict[str, Any]:
    """Build an AppEEARS point task payload.

    Args:
        task_name: AppEEARS task name.
        coordinates: One ``(longitude, latitude)`` pair or AppEEARS coordinate
            dictionaries containing longitude and latitude.
        layers: Task-ready AppEEARS layers.
        start_date: Start date as ``YYYY-MM-DD`` or ``MM-DD-YYYY``.
        end_date: End date as ``YYYY-MM-DD`` or ``MM-DD-YYYY``.
        recurring: Whether the task uses recurring dates.
        year_range: Two-year range for recurring tasks.

    Returns:
        AppEEARS point task payload.
    """

    params = {
        "dates": [
            _date_range(
                start_date=start_date,
                end_date=end_date,
                recurring=recurring,
                year_range=year_range,
            )
        ],
        "layers": list(layers),
        "coordinates": _normalize_coordinates(coordinates),
    }
    return {"task_type": "point", "task_name": task_name, "params": params}

`appeears_products(keyword=None, platform=None, available=True, client=None)` ¶

List AppEEARS products.

Parameters:

Name	Type	Description	Default
`keyword`	`Optional[str]`	Case-insensitive search text matched against product fields.	`None`
`platform`	`Optional[str]`	Platform name, such as `"EMIT"`.	`None`
`available`	`Optional[bool]`	If set, filter by the AppEEARS availability flag.	`True`
`client`	`Optional[AppEEARSClient]`	Optional AppEEARS client.	`None`

Returns:

Type	Description
`List[Dict[str, Any]]`	List of product metadata dictionaries.

Source code in hypercoast/appeears.py

def appeears_products(
    keyword: Optional[str] = None,
    platform: Optional[str] = None,
    available: Optional[bool] = True,
    client: Optional[AppEEARSClient] = None,
) -> List[Dict[str, Any]]:
    """List AppEEARS products.

    Args:
        keyword: Case-insensitive search text matched against product fields.
        platform: Platform name, such as ``"EMIT"``.
        available: If set, filter by the AppEEARS availability flag.
        client: Optional AppEEARS client.

    Returns:
        List of product metadata dictionaries.
    """

    client = client or AppEEARSClient()
    return client.products(keyword=keyword, platform=platform, available=available)

`appeears_submit_task(task, client=None)` ¶

Submit an AppEEARS task.

Parameters:

Name	Type	Description	Default
`task`	`Dict[str, Any]`	AppEEARS task payload.	required
`client`	`Optional[AppEEARSClient]`	Authenticated AppEEARS client.	`None`

Returns:

Type	Description
`Dict[str, Any]`	AppEEARS task submission response.

Source code in hypercoast/appeears.py

def appeears_submit_task(
    task: Dict[str, Any],
    client: Optional[AppEEARSClient] = None,
) -> Dict[str, Any]:
    """Submit an AppEEARS task.

    Args:
        task: AppEEARS task payload.
        client: Authenticated AppEEARS client.

    Returns:
        AppEEARS task submission response.
    """

    if client is None:
        client = appeears_login()
    return client.submit_task(task)

`appeears_wait(task_id, client=None, interval=20, timeout=None, verbose=True)` ¶

Wait for an AppEEARS task to finish.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	AppEEARS task identifier.	required
`client`	`Optional[AppEEARSClient]`	Authenticated AppEEARS client.	`None`
`interval`	`int`	Polling interval in seconds.	`20`
`timeout`	`Optional[int]`	Optional timeout in seconds.	`None`
`verbose`	`bool`	Whether to print status changes.	`True`

Returns:

Type	Description
`Dict[str, Any]`	Final task metadata response.

Source code in hypercoast/appeears.py

def appeears_wait(
    task_id: str,
    client: Optional[AppEEARSClient] = None,
    interval: int = 20,
    timeout: Optional[int] = None,
    verbose: bool = True,
) -> Dict[str, Any]:
    """Wait for an AppEEARS task to finish.

    Args:
        task_id: AppEEARS task identifier.
        client: Authenticated AppEEARS client.
        interval: Polling interval in seconds.
        timeout: Optional timeout in seconds.
        verbose: Whether to print status changes.

    Returns:
        Final task metadata response.
    """

    if client is None:
        client = appeears_login()
    return client.wait_for_task(
        task_id=task_id, interval=interval, timeout=timeout, verbose=verbose
    )

`read_appeears(files, layers=None, product='EMIT_L2A_RFL.001', variable='reflectance', client=None, **kwargs)` ¶

Read AppEEARS raster output as an xarray hyperspectral cube.

Parameters:

Name	Type	Description	Default
`files`	`Union[str, os.PathLike[str], Sequence[Union[str, os.PathLike[str]]]]`	GeoTIFF or NetCDF path, directory, or sequence of paths.	required
`layers`	`Optional[Union[Dict[str, Dict[str, Any]], Sequence[Dict[str, Any]]]]`	Optional AppEEARS layer metadata from `appeears_layers` or `appeears_emit_layers(..., return_metadata=True)`.	`None`
`product`	`str`	AppEEARS product used to infer wavelength metadata when layers are omitted.	`'EMIT_L2A_RFL.001'`
`variable`	`str`	Output data variable name.	`'reflectance'`
`client`	`Optional[AppEEARSClient]`	Optional AppEEARS client used to fetch layer metadata.	`None`
`**kwargs`	`Any`	Additional keyword arguments passed to `rioxarray.open_rasterio`.	`{}`

Returns:

Type	Description
`Any`	xarray.Dataset containing the stacked raster data.

Source code in hypercoast/appeears.py

def read_appeears(
    files: Union[str, os.PathLike[str], Sequence[Union[str, os.PathLike[str]]]],
    layers: Optional[Union[Dict[str, Dict[str, Any]], Sequence[Dict[str, Any]]]] = None,
    product: str = EMIT_REFLECTANCE_PRODUCT,
    variable: str = "reflectance",
    client: Optional[AppEEARSClient] = None,
    **kwargs: Any,
) -> Any:
    """Read AppEEARS raster output as an xarray hyperspectral cube.

    Args:
        files: GeoTIFF or NetCDF path, directory, or sequence of paths.
        layers: Optional AppEEARS layer metadata from ``appeears_layers`` or
            ``appeears_emit_layers(..., return_metadata=True)``.
        product: AppEEARS product used to infer wavelength metadata when layers
            are omitted.
        variable: Output data variable name.
        client: Optional AppEEARS client used to fetch layer metadata.
        **kwargs: Additional keyword arguments passed to
            ``rioxarray.open_rasterio``.

    Returns:
        xarray.Dataset containing the stacked raster data.
    """

    paths = _coerce_paths(files)
    if layers is None:
        client = client or AppEEARSClient()
        layers = client.layers(product)
    metadata = _metadata_by_layer(product, layers)
    if paths and all(Path(path).suffix.lower() in {".nc", ".nc4"} for path in paths):
        return _read_appeears_netcdf(paths, metadata, product, variable, **kwargs)

    import rioxarray
    import xarray as xr

    arrays = []
    wavelengths = []
    band_names = []
    for path in paths:
        layer_name = _layer_from_filename(path)
        if layer_name is None:
            continue
        layer_info = metadata.get(layer_name)
        wavelength = None if layer_info is None else layer_info.get("wavelength")
        if wavelength is None:
            continue
        data = rioxarray.open_rasterio(path, masked=True, **kwargs)
        if "band" in data.dims and data.sizes.get("band") == 1:
            data = data.squeeze("band", drop=True)
        arrays.append(data)
        wavelengths.append(float(wavelength))
        band_names.append(layer_name)

    if not arrays:
        raise ValueError("No AppEEARS spectral GeoTIFF files could be read.")

    order = sorted(range(len(wavelengths)), key=lambda index: wavelengths[index])
    arrays = [arrays[index] for index in order]
    wavelengths = [wavelengths[index] for index in order]
    band_names = [band_names[index] for index in order]
    data = xr.concat(
        arrays,
        dim=xr.DataArray(wavelengths, dims="wavelength", name="wavelength"),
    )
    dataset = data.to_dataset(name=variable)
    dataset[variable].attrs["band_names"] = band_names
    dataset.attrs["source"] = "NASA AppEEARS"
    dataset.attrs["product"] = product
    return dataset

appeears module¶

AppEEARSClient ¶

headers: Dict[str, str] property readonly ¶

bundle(self, task_id) ¶

download_bundle(self, task_id, out_dir='.', file_ids=None, file_types=None, overwrite=False) ¶

download_file(self, task_id, file_id, file_name, out_dir='.', overwrite=False, chunk_size=8192) ¶

layers(self, product) ¶

login(self, username=None, password=None, token=None) ¶

products(self, keyword=None, platform=None, available=True) ¶

projections(self) ¶

s3credentials(self) ¶

spectral_layers(self, product='EMIT_L2A_RFL.001', wavelengths=None, include_flags=False) ¶

status(self, task_id) ¶

submit_task(self, task) ¶

task(self, task_id) ¶

tasks(self, limit=None, offset=None, pretty=False) ¶

wait_for_task(self, task_id, interval=20, timeout=None, verbose=True) ¶

appeears_area_task(task_name, geometry, layers, start_date, end_date, output_format=None, projection='geographic', recurring=False, year_range=None, orthorectify=True, additional_options=None) ¶

appeears_download(task_id, out_dir='.', client=None, file_types=None, overwrite=False) ¶

appeears_emit_layers(wavelengths=None, product='EMIT_L2A_RFL.001', include_flags=False, return_metadata=False, client=None) ¶

appeears_layers(product, client=None) ¶

appeears_login(username=None, password=None, token=None, api_url='https://appeears.earthdatacloud.nasa.gov/api/') ¶

appeears_point_task(task_name, coordinates, layers, start_date, end_date, recurring=False, year_range=None) ¶

appeears_products(keyword=None, platform=None, available=True, client=None) ¶

appeears_submit_task(task, client=None) ¶

appeears_wait(task_id, client=None, interval=20, timeout=None, verbose=True) ¶

read_appeears(files, layers=None, product='EMIT_L2A_RFL.001', variable='reflectance', client=None, **kwargs) ¶

`AppEEARSClient` ¶

`headers: Dict[str, str]` `property` `readonly` ¶

`bundle(self, task_id)` ¶

`download_bundle(self, task_id, out_dir='.', file_ids=None, file_types=None, overwrite=False)` ¶

`download_file(self, task_id, file_id, file_name, out_dir='.', overwrite=False, chunk_size=8192)` ¶

`layers(self, product)` ¶

`login(self, username=None, password=None, token=None)` ¶

`products(self, keyword=None, platform=None, available=True)` ¶

`projections(self)` ¶

`s3credentials(self)` ¶

`spectral_layers(self, product='EMIT_L2A_RFL.001', wavelengths=None, include_flags=False)` ¶

`status(self, task_id)` ¶

`submit_task(self, task)` ¶

`task(self, task_id)` ¶

`tasks(self, limit=None, offset=None, pretty=False)` ¶

`wait_for_task(self, task_id, interval=20, timeout=None, verbose=True)` ¶

`appeears_area_task(task_name, geometry, layers, start_date, end_date, output_format=None, projection='geographic', recurring=False, year_range=None, orthorectify=True, additional_options=None)` ¶

`appeears_download(task_id, out_dir='.', client=None, file_types=None, overwrite=False)` ¶

`appeears_emit_layers(wavelengths=None, product='EMIT_L2A_RFL.001', include_flags=False, return_metadata=False, client=None)` ¶

`appeears_layers(product, client=None)` ¶

`appeears_login(username=None, password=None, token=None, api_url='https://appeears.earthdatacloud.nasa.gov/api/')` ¶

`appeears_point_task(task_name, coordinates, layers, start_date, end_date, recurring=False, year_range=None)` ¶

`appeears_products(keyword=None, platform=None, available=True, client=None)` ¶

`appeears_submit_task(task, client=None)` ¶

`appeears_wait(task_id, client=None, interval=20, timeout=None, verbose=True)` ¶

`read_appeears(files, layers=None, product='EMIT_L2A_RFL.001', variable='reflectance', client=None, **kwargs)` ¶