process.py

"""
Implementation of `Sentinel Hub Process API interface <https://docs.sentinel-hub.com/api/latest/api/process/>`__.
"""

from __future__ import annotations

from typing import Any, Iterable

import requests

from sentinelhub.exceptions import DownloadFailedException

from ..config import SHConfig
from ..constants import MimeType
from ..download import SentinelHubDownloadClient
from ..geometry import BBox, Geometry
from ..types import JsonDict
from .base_request import InputDataDict, SentinelHubBaseApiRequest
from .utils import AccessSpecification, _update_other_args, remove_undefined, s3_specification


class SentinelHubRequest(SentinelHubBaseApiRequest):
    """Sentinel Hub Process API interface

    For more information check
    `Process API documentation <https://docs.sentinel-hub.com/api/latest/api/process/>`__.
    """

    _SERVICE_ENDPOINT = "process"

    def __init__(
        self,
        evalscript: str,
        input_data: list[JsonDict | InputDataDict],
        responses: list[JsonDict],
        bbox: BBox | None = None,
        geometry: Geometry | None = None,
        size: tuple[int, int] | None = None,
        resolution: tuple[float, float] | None = None,
        **kwargs: Any,
    ):
        """
        For details of certain parameters check the
        `Process API reference <https://docs.sentinel-hub.com/api/latest/reference/#operation/process>`_.

        :param evalscript: `Evalscript <https://docs.sentinel-hub.com/api/latest/#/Evalscript/>`__.
        :param input_data: A list of input dictionary objects as described in the API reference. It can be generated
            with the helper method `SentinelHubRequest.input_data`
        :param responses: A list of `output.responses` objects as described in the API reference. It can be generated
            with the helper function `SentinelHubRequest.output_response`
        :param bbox: Bounding box describing the area of interest.
        :param geometry: Geometry describing the area of interest.
        :param size: Size of the image.
        :param resolution: Resolution of the image. It has to be in units compatible with the given CRS.
        :param data_folder: location of the directory where the fetched data will be saved.
        :param config: A custom instance of config class to override parameters from the saved configuration.
        """
        if not isinstance(evalscript, str):
            raise ValueError("'evalscript' should be a string")

        parsed_mime_type = MimeType.from_string(responses[0]["format"]["type"])
        self._mime_type = MimeType.TAR if len(responses) > 1 else parsed_mime_type

        self.payload = self.body(
            request_bounds=self.bounds(bbox=bbox, geometry=geometry),
            request_data=input_data,
            request_output=self.output(size=size, resolution=resolution, responses=responses),
            evalscript=evalscript,
        )

        super().__init__(SentinelHubDownloadClient, **kwargs)

    @property
    def mime_type(self) -> MimeType:
        return self._mime_type

    @staticmethod
    def body(
        request_bounds: JsonDict,
        request_data: list[JsonDict],
        evalscript: str,
        request_output: JsonDict | None = None,
        other_args: JsonDict | None = None,
    ) -> JsonDict:
        """Generate the Process API request body

        :param request_bounds: A dictionary as generated by `SentinelHubRequest.bounds` helper method.
        :param request_data: A list of dictionaries as generated by `SentinelHubRequest.input_data` helper method.
        :param evalscript: `Evalscript <https://docs.sentinel-hub.com/api/latest/#/Evalscript/>`__.
        :param request_output: A dictionary as generated by `SentinelHubRequest.output` helper method.
        :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
            by it.
        """
        request_body = {"input": {"bounds": request_bounds, "data": request_data}, "evalscript": evalscript}

        if request_output is not None:
            request_body["output"] = request_output

        if other_args:
            _update_other_args(request_body, other_args)

        return request_body

    @staticmethod
    def output_response(
        identifier: str, response_format: str | MimeType, other_args: JsonDict | None = None
    ) -> JsonDict:
        """Generate an element of `output.responses` as described in the Process API reference.

        :param identifier: Identifier of the output response.
        :param response_format: A mime type of one of 'png', 'json', 'jpeg', 'tiff'.
        :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
            by it.
        """
        output_response = {"identifier": identifier, "format": {"type": MimeType(response_format).get_string()}}

        if other_args:
            _update_other_args(output_response, other_args)

        return output_response

    @staticmethod
    def output(
        responses: list[JsonDict],
        size: tuple[int, int] | None = None,
        resolution: tuple[float, float] | None = None,
        other_args: dict | None = None,
    ) -> JsonDict:
        """Generate an `output` part of the request as described in the Process API reference

        :param responses: A list of objects in `output.responses` as generated by `SentinelHubRequest.output_response`.
        :param size: Size of the image.
        :param resolution: Resolution of the image. It has to be in units compatible with the given CRS.
        :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
            by it.
        """
        if size and resolution:
            raise ValueError("Either size or resolution argument should be given, not both.")

        request_output: JsonDict = {"responses": responses}

        if size:
            request_output["width"], request_output["height"] = size
        if resolution:
            request_output["resx"], request_output["resy"] = resolution

        if other_args:
            _update_other_args(request_output, other_args)

        return request_output


class AsyncProcessRequest(SentinelHubBaseApiRequest):
    """Sentinel Hub Async Process API interface

    For more information check
    `Async Process API documentation <https://docs.sentinel-hub.com/api/latest/api/async-process/>`__.
    """

    _SERVICE_ENDPOINT = "async/process"

    def __init__(
        self,
        *,
        evalscript: str | None = None,
        evalscript_reference: AccessSpecification | None = None,
        input_data: list[JsonDict | InputDataDict],
        responses: list[JsonDict],
        delivery: AccessSpecification,
        bbox: BBox | None = None,
        geometry: Geometry | None = None,
        size: tuple[int, int] | None = None,
        resolution: tuple[float, float] | None = None,
        **kwargs: Any,
    ):
        """
        For details of certain parameters check the
        `Async Process API reference <https://docs.sentinel-hub.com/api/latest/reference/#tag/async_process>`_.

        :param evalscript: `Evalscript <https://docs.sentinel-hub.com/api/latest/#/Evalscript/>`__.
        :param evalscript_reference:
            `Evalscript specification <https://docs.sentinel-hub.com/api/latest/reference/#tag/async_process>`__.
        :param input_data: A list of input dictionary objects as described in the API reference. It can be generated
            with the helper method `SentinelHubRequest.input_data`
        :param responses: A list of `output.responses` objects as described in the API reference. It can be generated
            with the helper function `SentinelHubRequest.output_response`
        :param delivery: S3 specification for dumping the request data.
        :param bbox: Bounding box describing the area of interest.
        :param geometry: Geometry describing the area of interest.
        :param size: Size of the image.
        :param resolution: Resolution of the image. It has to be in units compatible with the given CRS.
        :param data_folder: location of the directory where the fetched data will be saved.
        :param config: A custom instance of config class to override parameters from the saved configuration.
        """

        self._mime_type = MimeType.JSON

        self.payload = self.body(
            request_bounds=self.bounds(bbox=bbox, geometry=geometry),
            request_data=input_data,
            request_output=self.output(size=size, resolution=resolution, responses=responses, delivery=delivery),
            evalscript=evalscript,
            evalscript_reference=evalscript_reference,
        )

        super().__init__(SentinelHubDownloadClient, **kwargs)

    s3_specification = s3_specification

    @property
    def mime_type(self) -> MimeType:
        return self._mime_type

    @staticmethod
    def body(
        request_bounds: JsonDict,
        request_data: list[JsonDict],
        evalscript: str | None,
        evalscript_reference: AccessSpecification | None,
        request_output: JsonDict | None = None,
        other_args: JsonDict | None = None,
    ) -> JsonDict:
        """Generate the Async Process API request body

        :param request_bounds: A dictionary as generated by `SentinelHubRequest.bounds` helper method.
        :param request_data: A list of dictionaries as generated by `SentinelHubRequest.input_data` helper method.
        :param evalscript: `Evalscript <https://docs.sentinel-hub.com/api/latest/#/Evalscript/>`__.
        :param request_output: A dictionary as generated by `SentinelHubRequest.output` helper method.
        :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
            by it.
        """
        request_body = {
            "input": {"bounds": request_bounds, "data": request_data},
            "evalscript": evalscript,
            "evalscriptReference": evalscript_reference,
        }

        if request_output is not None:
            request_body["output"] = request_output

        if other_args:
            _update_other_args(request_body, other_args)

        return remove_undefined(request_body)

    @staticmethod
    def output_response(
        identifier: str, response_format: str | MimeType, other_args: JsonDict | None = None
    ) -> JsonDict:
        """Generate an element of `output.responses` as described in the Async Process API reference.

        :param identifier: Identifier of the output response.
        :param response_format: A mime type of one of 'png', 'json', 'jpeg', 'tiff'.
        :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
            by it.
        """
        output_response = {"identifier": identifier, "format": {"type": MimeType(response_format).get_string()}}

        if other_args:
            _update_other_args(output_response, other_args)

        return output_response

    @staticmethod
    def output(
        responses: list[JsonDict],
        delivery: AccessSpecification,
        size: tuple[int, int] | None = None,
        resolution: tuple[float, float] | None = None,
        other_args: dict | None = None,
    ) -> JsonDict:
        """Generate an `output` part of the request as described in the Async Process API reference

        :param responses: A list of objects in `output.responses` as generated by `AsyncProcessRequest.output_response`.
        :param size: Size of the image.
        :param resolution: Resolution of the image. It has to be in units compatible with the given CRS.
        :param other_args: Additional dictionary of arguments. If provided, the resulting dictionary will get updated
            by it.
        """
        if size and resolution:
            raise ValueError("Either size or resolution argument should be given, not both.")

        request_output: JsonDict = {"responses": responses, "delivery": delivery}

        if size:
            request_output["width"], request_output["height"] = size
        if resolution:
            request_output["resx"], request_output["resy"] = resolution

        if other_args:
            _update_other_args(request_output, other_args)

        return request_output


def get_async_running_status(ids: Iterable[str], config: SHConfig | None = None) -> dict[str, bool]:
    """Returns a mapping that describes which requests are running.

    :param ids: A collection of async request IDs.
    :param config: A custom instance of config class to override parameters from the saved configuration.
    :return: A mapping that specifies whether a process is running for each of the IDs.
    """
    config = config or SHConfig()
    client = SentinelHubDownloadClient(config=config)
    result = {}
    for request_id in ids:
        try:
            client.get_json_dict(f"{config.sh_base_url}/api/v1/async/process/{request_id}", use_session=True)
            # A successful request means it's running
            result[request_id] = True
        except DownloadFailedException as exception:  # noqa: PERF203
            # A 404 means it's not running
            if exception.request_exception is not None and exception.request_exception.response is not None:
                if exception.request_exception.response.status_code == requests.status_codes.codes.NOT_FOUND:
                    result[request_id] = False
            else:
                raise exception from exception

    return result