cso_dataspace.py

#
# Changes
#
# 2023-10, Arjo Segers
#   Tools to access Copernicus DataSpace.
#

########################################################################
###
### help
###
########################################################################

"""
.. _cso-dataspace:

*************
CSO DataSpace
*************

The ``cso_dataspace`` module provides classes for accessing  data from the 
`Copernicus DataSpace <https://dataspace.copernicus.eu/>`_.

To browse through the data, use the `Browser <https://dataspace.copernicus.eu/browser/>`_.


.. _dataspace-account:

Account setup
=============

To be able to download data from the *DataSpace*, first 
`Register and create an account <https://documentation.dataspace.copernicus.eu/Registration.html>`_.

On a Linux system, login/passwords for websites can be stored in the users ``.netrc`` file
in the home directory. Create this file if it does not exist yet, and add the following 
line with the login name of the account (your email) and the chosen password::

    machine zipper.dataspace.copernicus.eu      login Your.Name@institute.org   password ***********

The file should be readible and writable for you only::

    chmod 400 ~/.netrc


.. _dataspace-api:

DataSpace API's
===============

The *DataSpace* could be access with a number of different 
`APIs <https://documentation.dataspace.copernicus.eu/APIs.html>`_.
    
Currently the `OpenSearch API <https://documentation.dataspace.copernicus.eu/APIs/OpenSearch.html>`_
is used as that was the first that worked as needed.

In future the `STAC API <https://stacspec.org/>`_ might be used, 
as this is becoming more and more the standard in the Earth Observation community.
Within CSO it was already used by for example :ref:`pal-api`, but could not get working yet
for the *DataSpace*.
See the `STAC product catalog <https://documentation.dataspace.copernicus.eu/APIs/STAC.html>`_
for more information.


Class hierchy
=============

The classes and are defined according to the following hierchy:


* :py:class:`.UtopyaRc`

  * :py:class:`.CSO_DataSpace_Inquire`

* :py:class:`CSO_DataSpace_DownloadFile`
* :py:class:`NullAuth`


Classes
=======


"""


########################################################################
###
### modules
###
########################################################################

# modules:
import logging
import requests

# tools:
import utopya


########################################################################
###
### OpenSearch inquire
###
########################################################################


class CSO_DataSpace_Inquire(utopya.UtopyaRc):

    """
    Inquire available Sentinel data from the
    `Copernicus DataSpace <https://dataspace.copernicus.eu/>`_.
    
    Before data could be downloaded from the *DataSpace*, setup your :ref:`dataspace-account`.
    
    Currently the `OpenSearch API <https://documentation.dataspace.copernicus.eu/APIs/OpenSearch.html>`_
    is used as that was the first that worked as needed;
    in future, the `STAC product catalog <https://documentation.dataspace.copernicus.eu/APIs/STAC.html>`_
    might be used.

    A query is sent to search for products that are available 
    for a certain time and overlap with a specified region. 
    The result is a list with orbit files and instructions on how to download them.
    
    In the settings, specify the time range over which files should be downloaded::

      <rcbase>.timerange.start  :  2018-07-01 00:00
      <rcbase>.timerange.end    :  2018-07-01 23:59

    Specify the base url of the API::

      <rcbase>.url              :  https://finder.creodias.eu/resto/api
      
    Define the collection name with::
    
      <rcbase>.collection       :  Sentinel5P

    Provide a product type::

        ! product type (always 10 characters!):
        <rcbase>.producttype            :  L2__NO2___
    
    Eventually specify a target area, only orbits with some pixels within the defined box will be downloaded::

        ! target area, leave empty for globe; format:  west,south,east,north
        <rcbase>.area             :  
        !<rcbase>.area             :  -30,30,35,76
      
    The table will also create the url's to download a file;
    specifity the template that should be used:

        ! template for download url given "{product_id}":
        <rcbase>.download_url     :  https://zipper.dataspace.copernicus.eu/odata/v1/Products({product_id})/$value
      
    Name of output csv file::
    
        ! output table, here including date of today:
        <rcbase>.output.file            :  ${my.work}/PAL_S5P_NO2_%Y-%m-%d.csv
    
    Example records (with extra whitespace to show the columns)::
    
        orbit;start_time         ;end_time           ;processing;collection;processor_version;filename                                                                              ;href
        11488;2020-01-01 02:34:16;2020-01-01 04:15:46;RPRO      ;03        ;020400           ;S5P_RPRO_L2__CH4____20200101T023416_20200101T041546_11488_03_020400_20221120T003820.nc;https://zipper.dataspace.copernicus.eu/odata/v1/Products(b3f240e6-505d-4cae-97ea-43a8778a318d)/$value
        11487;2020-01-01 00:52:46;2020-01-01 02:34:16;RPRO      ;03        ;020400           ;S5P_RPRO_L2__CH4____20200101T005246_20200101T023416_11487_03_020400_20221120T003818.nc;https://zipper.dataspace.copernicus.eu/odata/v1/Products(a3d40f81-6c86-44bc-bc4b-457ff069b121)/$value
        :

    """

    def __init__(self, rcfile, rcbase="", env={}, indent=""):

        """
        Inquire oribt files.
        """

        # modules:
        import sys
        import os
        import datetime
        import calendar
        import time
        import requests
        import pandas

        # info ...
        logging.info(f"{indent}")
        logging.info(f"{indent}** Inquire files available on Copernicus DataSpace")
        logging.info(f"{indent}")

        # init base object:
        utopya.UtopyaRc.__init__(self, rcfile=rcfile, rcbase=rcbase, env=env)

        # url of API:
        api_url = self.GetSetting("url")
        # info ...
        logging.info(f"{indent}API url      : {api_url}")

        # template url for downloads:
        download_url = self.GetSetting("download_url")
        # info ...
        logging.info(f"{indent}download url : {download_url}")

        # collection:
        collection = self.GetSetting("collection")
        # info ...
        logging.info(f"{indent}collection   : {collection}")

        # combine into search url:
        search_url = f"{api_url}/collections/{collection}/search.json"

        ## authorization is done by header dict:.
        # headers = { "Authorization" : f"access_token {access_token}" }

        # time range:
        t1 = self.GetSetting("timerange.start", totype="datetime")
        t2 = self.GetSetting("timerange.end", totype="datetime")
        # info ...
        tfmt = "%Y-%m-%d %H:%M"
        logging.info(f"{indent}timerange    : [{t1.strftime(tfmt)},{t2.strftime(tfmt)}")

        # product type (always 10 characters!):
        #    L2__NO2___
        producttype = self.GetSetting("producttype")
        # info ...
        logging.info(f"{indent}product type : {producttype}")

        # area of interest: west,south:east,north
        area = self.GetSetting("area")
        # defined?
        if len(area) > 0:
            # convert from format for "dhusget.sh":
            #    west,south:east,north
            west, south, east, north = map(float, area.replace(":", " ").replace(",", " ").split())
            # info ...
            logging.info(
                f"{indent}area         : [{west:.2f},{east:.2f}] x [{south:.2f},{north:.2f}]"
            )
            # box parameter:
            box = f"{west},{east},{south},{north}"
        else:
            # info ...
            logging.info(f"{indent}area         : no")
            # box parameter:
            box = None
        # endif

        # target file, might include time templates:
        output_file__template = self.GetSetting("output.file")
        # current time:
        output_file = datetime.datetime.now().strftime(output_file__template)

        # initialize output table:
        output_df = pandas.DataFrame()

        # info ...
        logging.info(f"{indent}search all items in timerange ...")

        # search query could only return a maximum number of records;
        # a 'page' of records is requested using a row offset and the number of rows:
        row0 = 0
        nrow = 100

        # initialize search parameters;
        # for possible content, see:
        #   https://documentation.dataspace.copernicus.eu/APIs/OpenSearch.html
        params = {}
        # fill maximum time range:
        tfmt = "%Y-%m-%dT%H:%M:%SZ"
        params["startDate"] = t1.strftime(tfmt)
        params["completionDate"] = t2.strftime(tfmt)
        if box is not None:
            params["box"] = box
        # endif
        # fill product type:
        params["productType"] = producttype
        # fill paging info:
        params["maxRecords"] = nrow

        # init counter:
        ipage = 0

        # loop over pages of query result:
        while True:

            # increase counter:
            ipage += 1
            # info ...
            logging.info(f"{indent}  page {ipage} (entries {row0+1},..,{row0+nrow})")

            # fill page number:
            params["page"] = ipage

            # number of tries:
            ntry = 1
            maxtry = 5
            # repeat a few times if necessary:
            while ntry <= maxtry:
                # send query to search page; no authorization is needed ...
                r = requests.get(search_url, params=params)
                # check status, raise error if request failed:
                try:
                    r.raise_for_status()
                except Exception as err:
                    msg = str(err)
                    logging.error(f"{indent}    from query; message received:")
                    logging.error(f"{indent}      %s" % msg)
                    if ntry == maxtry:
                        logging.error(f"{indent}    tried {ntry} times now, exit ...")
                        raise Exception
                    else:
                        logging.error(f"{indent}    wait ..")
                        time.sleep(10)
                        logging.error(f"{indent}    try again ...")
                        ntry += 1
                        continue
                    # endif
                # endtry
                # no error, leave:
                break
            # endwhile

            # While testing: save the result as a json file, and load it into a browser.
            # This shows a dict with among others the fields:
            #
            #   { ..
            #       'features' : [          # list of orbits, in browser named: '0','1',...
            #           { 'id' : '0f318743-8bb9-55ed-b42d-7721b24f7ede',   # download id
            #             'properties' : {
            #                 'title'  : "S5P_OFFL_L2__CH4____20220531T224613_20220601T002743_23999_02_020301_20220602T143707.nc",
            #                 ...
            #                 }
            #             ...
            #           },
            #           ...
            #       ]
            #  }
            #
            # save result?
            if True:
                # targefile:
                qfile = "query.json"
                # save:
                with open(qfile, "w") as f:
                    f.write(r.text)
                # endwith
            # endif

            # convert response to json dict:
            data = r.json()

            # check ...
            if type(data) != dict:
                logging.error(f"request response should be a json dict, found type: {type(data)}")
                raise Exception
            # endif
            # check ...
            if "features" not in data.keys():
                logging.error(f"element 'features' not found in response")
                raise Exception
            # endif

            # count:
            nrec = len(data["features"])
            # loop over features:
            for feature in data["features"]:

                # check ...
                if type(feature) != dict:
                    logging.error(f"feature should be a dict, found type: {type(feature)}")
                    raise Exception
                # endif

                # check ...
                if "id" not in feature.keys():
                    logging.error(f"element 'id' not found in feature")
                    raise Exception
                # endif
                # get product id:
                product_id = feature["id"]

                # check ...
                if "properties" not in feature.keys():
                    logging.error(f"element 'properties' not found in feature")
                    raise Exception
                # endif
                # check ...
                if "title" not in feature["properties"].keys():
                    logging.error(f"element 'properties/title' not found in feature")
                    raise Exception
                # endif
                # get full filename:
                filename = feature["properties"]["title"]
                #
                #    S5P_OFFL_L2__NO2____20180701T005930_20180701T024100_03698_01_010002_20180707T022838.nc
                #    plt proc [product-] [starttime....] [endtime......] orbit cl procrv [prodtime.....]
                #
                bname = os.path.basename(filename).replace(".nc", "")
                # split:
                platform_name, processing, rest = bname.split("_", 2)
                product_type = rest[0:10]
                parts = rest[11:].split("_")
                start_time, end_time, orbit, collection, processor_version, prod_time = parts

                # convert:
                tfmt = "%Y%m%dT%H%M%S"
                ts = datetime.datetime.strptime(start_time, tfmt)
                te = datetime.datetime.strptime(end_time, tfmt)

                # fill download href:
                href = download_url.format(product_id=product_id)

                # strange, sometimes records seem double ...
                # already records present?
                if len(output_df) > 0:
                    # same href already stored?
                    if href in output_df["href"].values:
                        ## testing ...
                        # logging.warning(f"ignore double product_id: {product_id}")
                        # ignore record:
                        continue
                    # endif
                # endif

                # fill record, values should be lists for concatenation below:
                rec = {
                    "orbit": [orbit],
                    "start_time": [ts],
                    "end_time": [te],
                    "processing": [processing],
                    "collection": [collection],
                    "processor_version": [processor_version],
                    "filename": [filename],
                    "href": [href],
                }
                # add record:
                output_df = pandas.concat((output_df, pandas.DataFrame(rec)), ignore_index=True)

            # endfor features

            ## testing...
            # if ipage == 9 :
            #    logging.warning( f"break after page {ipage} ..." )
            #    break
            ## endif

            # not a full page? then end is reached ...
            if nrec < nrow:
                # leave loop over pages:
                break
            # endif

            # increse row offset:
            row0 += nrow

        # endwhile # pages

        # info ..
        logging.info("save to: %s ..." % output_file)
        # create directory:
        dirname = os.path.dirname(output_file)
        if len(dirname) > 0:
            if not os.path.isdir(dirname):
                os.makedirs(dirname)
            # endif
        # endif
        # write:
        output_df.to_csv(output_file, sep=";", index=False)

        # info ...
        logging.info(f"{indent}")
        logging.info(f"{indent}** end inquire")
        logging.info(f"{indent}")

    # enddef __init__


# endclass CSO_DataSpace_Inquire


########################################################################
###
### OpenSearch download
###
########################################################################


class NullAuth(requests.auth.AuthBase):

    """
    Force requests to ignore the ``~/.netrc`` file.

    Some sites do not support regular authentication, but we still
    want to store credentials in the ``~/.netrc`` file and submit them
    as form elements. Without this, requests would otherwise use the
    ``~/.netrc`` which leads, on some sites, to a 401 error.

    Use with::

        requests.get( url, auth=NullAuth() )
        
    Source:
    
      `<https://github.com/psf/requests/issues/2773#issuecomment-174312831>`_

    """

    def __call__(self, r):
        return r

    # enddef __call__


# endclass NullAuth


# *


class CSO_DataSpace_DownloadFile(object):

    """
    Download single file from *Copernicus DataSpace*.
    
    Arguments:

    * ``href`` : download url, for example::
    
        https://zipper.dataspace.copernicus.eu/odata/v1/Products('d483baa0-3a61-4985-aa0c-5642a83c9214')/$value

    * ``output_file`` : target file
    
    Optional arguments:
    
    * ``maxtry`` : number of times to try again if download fails
    * ``timeout`` : delay in seconds between requests
    """

    def __init__(self, href, output_file, maxtry=10, timeout=60, indent=""):

        """
        Download file.
        """

        # modules:
        import os
        import urllib.parse
        import requests
        import zipfile
        import shutil

        # tools:
        import cso_file

        #
        # On linux system, login/passwords for websites and ftp can be stored in "~/.netrc" file:
        #   ---[~/.netrc]-----------------------------------------------
        #   machine zipper.dataspace.copernicus.eu      login Your.Name@institute.org   password ***********
        #   ------------------------------------------------------------
        # Retrieve the login/password from ~/.netrc to avoid hardcoding them in a script.
        #
        # the "get_netrc_auth" function requires base of url as first argument,
        # for example:  https://zipper.dataspace.copernicus.eu
        # extract parts from download url:
        p = urllib.parse.urlparse(href)
        url = f"{p.scheme}://{p.netloc}"
        # get username and password from ~/.netrc file:
        try:
            username, password = requests.utils.get_netrc_auth(url, raise_errors=True)
        except:
            logging.error(f"Could not get username and password from ~/.netrc file for url:")
            logging.error(f"  {url}")
            logging.error(f"For the Copernicus DataSpace, the file should contain:")
            logging.error(f"  machine {p.netloc}   login ****  password ****")
            raise Exception
        # endtry

        # convert into token for dataspace website following:
        #   https://documentation.dataspace.copernicus.eu/APIs/Token.html
        # fill data fields:
        data = {
            "client_id": "cdse-public",
            "username": username,
            "password": password,
            "grant_type": "password",
        }
        # identity server:
        domain = "identity.dataspace.copernicus.eu"
        url = f"https://{domain}/auth/realms/CDSE/protocol/openid-connect/token"
        try:
            # send request:
            r = requests.post(url, data=data)
            # check status, raise error if request failed:
            r.raise_for_status()
        except requests.exceptions.HTTPError as err:
            # info ..
            msg = str(err)
            logging.error(f"exception from download; message received:")
            logging.error(f"  {msg}")
            # catch known problem ...
            if msg.startswith("401 Client Error: Unauthorized for url:"):
                logging.error(f"Interpretation: the (username,password) received from")
                logging.error(f"your '~/.netrc' file are incorrect.")
                logging.error(f"For the Copernicus DataSpace, the file should contain:")
                logging.error(f"  machine {p.netloc}   login ****  password ****")
                logging.error(f"If the machine was not found, a default might have been received")
                raise Exception
            else:
                raise Exception(f"Access token creation failed; server response: {r.json()}")
            # endif
        except:
            raise Exception(f"Access token creation failed; server response: {r.json()}")
        # endtry # get access token
        # extract token from response:
        access_token = r.json()["access_token"]

        # retry loop ..
        ntry = 0
        while True:
            # try to download and save:
            try:

                # try to download:
                try:

                    # fill authorization token in header:
                    headers = {"Authorization": f"Bearer {access_token}"}
                    # ensure that "~/.netrc" is ignored by passing null-authorization,
                    # otherwise the token in the header is overwritten by a token formed
                    # from the login/password in the rcfile if that is found:
                    r = requests.get(href, auth=NullAuth(), headers=headers, timeout=timeout)
                    # check status, raise error if request failed:
                    r.raise_for_status()

                    # product is a zip-file:
                    product_file = "product.zip"
                    # info ..
                    logging.info(f"{indent}      write to {product_file} ...")
                    # write to temporary target first ..
                    tmpfile = product_file + ".tmp"
                    # open destination file for binary write:
                    with open(tmpfile, "wb") as fd:
                        # prefered way to write content following:
                        #   https://docs.python-requests.org/en/master/user/quickstart/
                        for chunk in r.iter_content(chunk_size=128):
                            fd.write(chunk)
                        # endfor
                    # endwith
                    # rename:
                    os.rename(tmpfile, product_file)

                    # open product file:
                    arch = zipfile.ZipFile(product_file, mode="r")
                    # loop over members, probably two files in a directory:
                    #   S5P_RPRO_L2__CH4____20200101T005246_etc/S5P_RPRO_L2__CH4____20200101T005246_etc.cdl
                    #                                           S5P_RPRO_L2__CH4____20200101T005246_etc.nc
                    for member in arch.namelist():
                        # ncfile?
                        if member.endswith(".nc"):
                            # this should be the target file ..
                            if os.path.basename(member) != os.path.basename(output_file):
                                logging.error(f"member of archive file: {member}")
                                logging.error(f"differs from target name: {output_file}")
                                raise Exception
                            # endif
                            # info ..
                            logging.info(f"{indent}      extract {member} ...")
                            # extract here, including leading directory:
                            arch.extract(member)
                            # info ..
                            logging.info(f"{indent}      store ...")
                            # create target dir if necessary:
                            cso_file.CheckDir(output_file)
                            # move to destination:
                            os.rename(member, output_file)
                            # remove directory tree:
                            shutil.rmtree(os.path.dirname(member))
                            # only one file in package; leave loop over members
                            break
                        # endif
                    # endfor # members
                    # info ..
                    logging.info(f"{indent}      remove product file ...")
                    # remove package:
                    os.remove(product_file)

                except requests.exceptions.HTTPError as err:
                    # info ..
                    msg = str(err)
                    logging.error("exception from download; message received:")
                    logging.error("  %s" % msg)

                except MemoryError as err:
                    logging.error("memory error from download; increase resources?")
                    # quit with error:
                    raise

                except Exception as err:
                    # info ..
                    logging.error("from download; message received:")
                    logging.error("  %s" % str(err))
                    # quit with error:
                    raise

                # endtry

            # error from download or save:
            except:
                # increase counter:
                ntry += 1
                # switch:
                if ntry == maxtry:
                    logging.warning(f"{indent}      tried {maxtry} times ...")
                    raise Exception
                else:
                    logging.warning(f"{indent}      exception from download; try again ...")
                    continue  # while-loop
                # endif
            # endtry
            # leave retry loop,
            # either because download was ok,
            # or because maximum number of retries was reached:
            break

        # endwhile # retry

    # enddef __init__


# endclass CSO_DataSpace_DownloadFile


########################################################################
###
### end
###
########################################################################