hda_api_functions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src='./img/opengeohub_logo.png' alt='Logo EU Copernicus EUMETSAT' align='right' width='20%'></img>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Harmonized Data Access API - Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook lists all `functions` that are defined and used to access data from `WEkEO` with the HDA API. You find the workflow describing how to access data with the HDA API [here](./12_ltpy_WEkEO_harmonized_data_access_api.ipynb). \n",
    "\n",
    "The following functions are available:\n",
    "- [generate_api_key](#generate_api_key)\n",
    "- [init](#init)\n",
    "- [get_access_token](#get_access_token)\n",
    "- [query_metadata](#query_metadata)\n",
    "- [accept_TandC](#accept_tc)\n",
    "- [get_job_id](#get_job_id)\n",
    "- [get_request_status](#get_request_status)\n",
    "- [get_results_list](#results_list)\n",
    "- [get_order_ids](#get_order_ids)\n",
    "- [get_order_status](#get_order_status)\n",
    "- [downloadFile](#download_file)\n",
    "- [get_filename_from_cd](#get_filename_from_cd)\n",
    "- [get_filenames](#get_filenames)\n",
    "- [download_data](#download_data)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Load required libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests, re, json\n",
    "import base64\n",
    "import shutil\n",
    "import time, os\n",
    "import urllib.parse"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `generate_api_key`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_api_key(username, password):\n",
    "    \"\"\" \n",
    "    Generates a Base64-encoded api key, based on the WEkEO user credentials username:password.\n",
    "    \n",
    "    Parameters:\n",
    "        username: WEkEO username\n",
    "        password: WEkEO password\n",
    "    \n",
    "    Returns:\n",
    "        Returns a string of the Base64-encoded api key\n",
    "    \"\"\"\n",
    "    s = username+':'+password\n",
    "    return (base64.b64encode(str.encode(s), altchars=None)).decode()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='init'></a> `init`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def init(dataset_id, api_key, download_dir_path):\n",
    "    \"\"\" \n",
    "    Initiates a dictionary with keys needed to use the HDA API.\n",
    "    \n",
    "    Parameters:\n",
    "        dataset_id: String representing the WEkEO collection id\n",
    "        api_key: Base64-encoded string\n",
    "        download_dir_path: directory path where data shall be downloaded to\n",
    "    \n",
    "    Returns:\n",
    "        Returns the initiated dictionary.\n",
    "    \"\"\"\n",
    "    hda_dict = {}\n",
    "    # Data broker address\n",
    "    hda_dict[\"broker_endpoint\"] = \"https://wekeo-broker.apps.mercator.dpi.wekeo.eu/databroker\"\n",
    "    # Terms and conditions\n",
    "    hda_dict[\"acceptTandC_address\"]\\\n",
    "            = hda_dict[\"broker_endpoint\"]\\\n",
    "            + \"/termsaccepted/Copernicus_General_License\"\n",
    "    # Access-token address\n",
    "    hda_dict[\"accessToken_address\"] = hda_dict[\"broker_endpoint\"]\\\n",
    "                                       + '/gettoken'\n",
    "    # Dataset id\n",
    "    hda_dict[\"dataset_id\"] = dataset_id\n",
    "    # API key\n",
    "    hda_dict[\"api_key\"] = api_key\n",
    "\n",
    "    # set HTTP success code\n",
    "    hda_dict[\"CONST_HTTP_SUCCESS_CODE\"] = 200\n",
    "\n",
    "    # download directory\n",
    "    hda_dict[\"download_dir_path\"] = download_dir_path\n",
    "    if not os.path.exists(download_dir_path):\n",
    "        os.makedirs(download_dir_path)\n",
    "\n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_access_token'></a> `get_access_token`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_access_token(hda_dict):\n",
    "    \"\"\" \n",
    "    Requests an access token to use the HDA API and stores it as separate key in the dictionary\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "    \n",
    "    Returns:\n",
    "        Returns the dictionary including the access token\n",
    "    \"\"\"\n",
    "    headers = { \n",
    "        'Authorization': 'Basic ' + hda_dict['api_key'] \n",
    "    }\n",
    "    data = [\n",
    "        ('grant_type', 'client_credentials'),\n",
    "    ]\n",
    "    print(\"Getting an access token. This token is valid for one hour only.\") \n",
    "    response = requests.get(hda_dict['accessToken_address'], headers=headers, verify=False)\n",
    "\n",
    "    # If the HTTP response code is 200 (i.e. success), then retrive the token from the response\n",
    "    if (response.status_code == hda_dict[\"CONST_HTTP_SUCCESS_CODE\"]):\n",
    "        access_token = json.loads(response.text)['access_token']\n",
    "        \n",
    "        print(\"Success: Access token is \" + access_token)\n",
    "    else:\n",
    "        print(\"Error: Unexpected response {}\".format(response))\n",
    "        print(response.headers)\n",
    "    \n",
    "    hda_dict['access_token'] = access_token\n",
    "    hda_dict['headers'] = {'Authorization': 'Bearer ' + hda_dict[\"access_token\"], 'Accept': 'application/json'}\n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='query_metadata'></a> `query_metadata`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def query_metadata(hda_dict):\n",
    "    \"\"\" \n",
    "    Requests metadata for the given dataset id and stores the response of the request in the dictionary.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "    \n",
    "    Returns:\n",
    "        Returns the dictionary including the query response\n",
    "    \"\"\"\n",
    "    response = requests.get(hda_dict['broker_endpoint'] + '/querymetadata/' + hda_dict['dataset_id'], headers=hda_dict['headers'])\n",
    "\n",
    "    print('Getting query metadata, URL Is ' + hda_dict['broker_endpoint'] + '/querymetadata/' + hda_dict['dataset_id'] +\"?access_token=\" + hda_dict['access_token'])\n",
    "    print(\"************** Query Metadata for \" + hda_dict['dataset_id'] +\" **************\") \n",
    "\n",
    "    if (response.status_code == hda_dict['CONST_HTTP_SUCCESS_CODE']):\n",
    "        parsedResponse = json.loads(response.text)\n",
    "        print(json.dumps(parsedResponse, indent=4, sort_keys=True))\n",
    "        print(\"**************************************************************************\")\n",
    "    else:\n",
    "        print(\"Error: Unexpected response {}\".format(response))\n",
    "    \n",
    "    hda_dict['parsedResponse']=parsedResponse\n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='accept_tc'></a> `acceptTandC`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def acceptTandC(hda_dict):\n",
    "    \"\"\" \n",
    "    Checks if the Terms and Conditions have been accepted and it not, they will be accepted. \\\n",
    "    The response is stored in the dictionary\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "    \n",
    "    Returns:\n",
    "        Returns the dictionary including the response from the Terms and Conditions statement.\n",
    "    \"\"\"\n",
    "    response = requests.get(hda_dict['acceptTandC_address'], headers=hda_dict['headers'])\n",
    "\n",
    "    isTandCAccepted = json.loads(response.text)['accepted']\n",
    "\n",
    "    if isTandCAccepted is False:\n",
    "        print(\"Accepting Terms and Conditions of Copernicus_General_License\")\n",
    "        response = requests.put(hda_dict['acceptTandC_address'], headers=hda_dict['headers'])\n",
    "    else:\n",
    "        print(\"Copernicus_General_License Terms and Conditions already accepted\")\n",
    "    isTandCAccepted = json.loads(response.text)['accepted']\n",
    "    hda_dict['isTandCAccepted']=isTandCAccepted\n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_job_id'></a> `get_job_id`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_job_id(hda_dict,data):\n",
    "    \"\"\" \n",
    "    Assigns a job id for the data request.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "        data: dictionary containing the dataset description\n",
    "\n",
    "    Returns:\n",
    "        Returns the dictionary including the assigned job id.\n",
    "    \"\"\"\n",
    "    response = requests.post(hda_dict['broker_endpoint'] + '/datarequest', headers=hda_dict['headers'], json=data, verify=False) \n",
    "\n",
    "    if (response.status_code == hda_dict['CONST_HTTP_SUCCESS_CODE']):\n",
    "        job_id=json.loads(response.text)['jobId']\n",
    "        print (\"Query successfully submitted. Job ID is \" + job_id)\n",
    "    else:\n",
    "        job_id=\"\"\n",
    "        print(\"Error: Unexpected response {}\".format(response))\n",
    "    \n",
    "    hda_dict['job_id']=job_id\n",
    "    get_request_status(hda_dict)\n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_request_status'></a> `get_request_status`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_request_status(hda_dict):\n",
    "    \"\"\" \n",
    "    Requests the status of the process to assign a job ID.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "    \"\"\"\n",
    "    status = \"not started\"\n",
    "    while (status != \"completed\"): \n",
    "        response = requests.get(hda_dict['broker_endpoint'] + '/datarequest/status/' + hda_dict['job_id'], headers=hda_dict['headers'])\n",
    "        if (response.status_code == hda_dict['CONST_HTTP_SUCCESS_CODE']):\n",
    "            status = json.loads(response.text)['status']\n",
    "            print (\"Query successfully submitted. Status is \" + status)\n",
    "        else:\n",
    "            print(\"Error: Unexpected response {}\".format(response))  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_results_list'></a> `get_results_list`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_results_list(hda_dict):\n",
    "    \"\"\" \n",
    "    Generates a list of filenames to be available for download\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "\n",
    "    Returns:\n",
    "        Returns the dictionary including the list of filenames to be downloaded.\n",
    "    \"\"\"\n",
    "    params = {'page':'0', 'size':'5'}\n",
    "    response = requests.get(hda_dict['broker_endpoint'] + '/datarequest/jobs/' + hda_dict['job_id'] + '/result', headers=hda_dict['headers'], params = params)\n",
    "    results = json.loads(response.text)\n",
    "    hda_dict['results']=results\n",
    "\n",
    "    print(\"************** Results *******************************n\")\n",
    "    print(json.dumps(results, indent=4, sort_keys=True))\n",
    "    print(\"*******************************************\")\n",
    "    \n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_order_ids'></a> `get_order_ids`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_order_ids(hda_dict):\n",
    "    \"\"\" \n",
    "    Assigns each file to be downloaded a unique order ID.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "\n",
    "    Returns:\n",
    "        Returns the dictionary including the list of order IDs and the request status of assigning the order IDs.\n",
    "    \"\"\"\n",
    "    i = 0\n",
    "    order_ids = []\n",
    "    order_sizes = []\n",
    "\n",
    "    for result in hda_dict['results']['content']:\n",
    "\n",
    "        data = {\n",
    "            \"jobId\": hda_dict['job_id'],\n",
    "            \"uri\": result['url']\n",
    "        }\n",
    "    \n",
    "        order_sizes.append(result['size'])\n",
    "\n",
    "        response = requests.post(hda_dict['broker_endpoint'] + '/dataorder', headers=hda_dict['headers'], json=data, verify=False)\n",
    "\n",
    "        if (response.status_code == hda_dict['CONST_HTTP_SUCCESS_CODE']):\n",
    "            order_ids.append(json.loads(response.text)['orderId'])\n",
    "            print (\"Query successfully submitted. Order ID is \" + order_ids[i])\n",
    "            response = get_order_status(hda_dict,order_ids[i])\n",
    "        else:\n",
    "            print(\"Error: Unexpected response {}\".format(response))\n",
    "        \n",
    "        i += 1\n",
    "    hda_dict['order_ids']=order_ids\n",
    "    hda_dict['order_sizes']=order_sizes\n",
    "    hda_dict['order_status_response']=response\n",
    "    return hda_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `get_order_status`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_order_status(hda_dict,order_id):\n",
    "    \"\"\" \n",
    "    Requests the status of assigning an order ID for a data file.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "        order_id: the order id for the data file\n",
    "\n",
    "    Returns:\n",
    "        Returns the response of assigning an order ID.\n",
    "    \"\"\"\n",
    "    status = \"not started\"\n",
    "    while (status != \"completed\"): \n",
    "        response = requests.get(hda_dict['broker_endpoint'] + '/dataorder/status/' + order_id, headers=hda_dict['headers'])\n",
    " \n",
    "        if (response.status_code == hda_dict['CONST_HTTP_SUCCESS_CODE']):\n",
    "            status = json.loads(response.text)['status']\n",
    "            print (\"Query successfully submitted. Status is \" + status)\n",
    "        else:\n",
    "             print(\"Error: Unexpected response {}\".format(response))\n",
    "    \n",
    "    return response"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='download_file'></a> `downloadFile`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def downloadFile(url, headers, directory, file_name, total_length = 0):\n",
    "    \"\"\" \n",
    "    Function to dowload a a single data file.\n",
    "    \n",
    "    Parameters:\n",
    "        url: is the download url which included the unique order ID\n",
    "        headers:\n",
    "        directory: download directory, where data file shall be stored\n",
    "        file_name: name of the data file\n",
    "        \n",
    "    Returns:\n",
    "        Returns the time needed to download the data file.\n",
    "    \"\"\"\n",
    "    r = requests.get(url, headers=headers, stream=True)\n",
    "    print('OK')\n",
    "    print('directory')\n",
    "    if r.status_code == 200:\n",
    "        filename = os.path.join(directory,  file_name)\n",
    "        print(\"Downloading \" + filename)\n",
    "        with open(filename, 'wb') as f:\n",
    "            start = time.process_time()\n",
    "            print(\"File size is: %8.2f MB\" % (total_length/(1024*1024)))\n",
    "            dl = 0\n",
    "            for chunk in r.iter_content(64738):\n",
    "                dl += len(chunk)\n",
    "                f.write(chunk)\n",
    "                if total_length is not None: # no content length header\n",
    "                    done = int(50 * dl / total_length)\n",
    "                    #sys.stdout.write(\"\\r[%s%s] %s kbps\" % ('=' * done, ' ' * (50-done), dl//(time.process_time() - start)))\n",
    "                    print(\"\\r[%s%s]  %8.2f Mbps\" % ('=' * done, ' ' * (50-done), (dl//(time.process_time() - start))/(1024*1024)), end='', flush=True)\n",
    "                    #print ('')\n",
    "                else:\n",
    "                    if( dl % (1024)  == 0 ):\n",
    "                        print(\"[%8.2f] MB downloaded, %8.2f kbps\" % (dl / (1024 * 1024), (dl//(time.process_time() - start))/1024))\n",
    "            print(\"[%8.2f] MB downloaded, %8.2f kbps\" % (dl / (1024 * 1024), (dl//(time.process_time() - start))/1024))\n",
    "            return (time.process_time() - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_filename_from_cd'></a> `get_filename_from_cd`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_filename_from_cd(cd):\n",
    "    \"\"\"\n",
    "    Get the filename from content disposition\n",
    "    \n",
    "    Parameters:\n",
    "        cd : content disposition (from https://www.w3.org/Protocols/rfc2616/rfc2616-sec19.html: \\\n",
    "            The Content-Disposition response-header field has been proposed as a means for the origin server to suggest a default filename\n",
    "            if the user requests that the content is saved to a file. This usage is derived from the definition of Content-Disposition in RFC 1806 [35].\n",
    "\n",
    "    Returns:\n",
    "        The filename from conten disposition\n",
    "    \"\"\"\n",
    "    if not cd:\n",
    "        return None\n",
    "    fname = re.findall('filename=(.+)', cd)\n",
    "    if len(fname) == 0:\n",
    "        return None\n",
    "    return fname[0][2:-1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='get_filenames'></a> `get_filenames`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_filenames(hda_dict):\n",
    "    \"\"\" \n",
    "    Generates a list of filenames taken from the results dictionary, retrieved with the function request_results_list.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "\n",
    "    Returns:\n",
    "        Returns a list of filenames for each entry stored in the dictionary returned by the function request_results_list.\n",
    "    \"\"\"\n",
    "    fileName = []\n",
    "    for file in hda_dict['results']['content']:\n",
    "        fileName.append(file['filename'])\n",
    "    return fileName"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='download_data'></a> `download_data`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def download_data(hda_dict):\n",
    "    \"\"\" \n",
    "    Downloads for each of the order IDs the associated data file.\n",
    "    \n",
    "    Parameters:\n",
    "        hda_dict: dictionary initied with the function init, that stores all required information to be able to interact with the HDA API\n",
    "    \"\"\"\n",
    "    fileName = get_filenames(hda_dict)\n",
    "    print(fileName)\n",
    "    i=0\n",
    "    for order_id in hda_dict['order_ids']:\n",
    "        print (order_id)\n",
    "        file_name=fileName[i]\n",
    "        \n",
    "        download_url = hda_dict['broker_endpoint'] + '/dataorder/download/' + order_id\n",
    "    \n",
    "        product_size = hda_dict['order_sizes'][i]\n",
    "        \n",
    "        print (download_url)\n",
    "        print (hda_dict['headers'])\n",
    "        print (type(hda_dict['download_dir_path']))\n",
    "        print (product_size)\n",
    "    \n",
    "        time_elapsed = downloadFile(download_url, hda_dict['headers'], hda_dict['download_dir_path'], file_name, product_size)\n",
    "    \n",
    "        print( \"Download complete...\")\n",
    "        print (\"Time Elapsed: \" + str(time_elapsed) + \" seconds\")\n",
    "        response=hda_dict['order_status_response']\n",
    "\n",
    "        if (response.status_code == hda_dict['CONST_HTTP_SUCCESS_CODE']):\n",
    "            order_file = \"./\" + file_name\n",
    "            if os.path.isfile(order_file):\n",
    "                print (\"Query successfully submitted. Response is \" + response)\n",
    "        else:\n",
    "            print(\"Error: Unexpected response {}\".format(response))\n",
    "        \n",
    "        i += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>\n",
    "&copy; 2020 | Julia Wagemann\n",
    "<a rel=\"license\" href=\"http://creativecommons.org/licenses/by/4.0/\"><img style=\"float: right\" alt=\"Creative Commons Lizenzvertrag\" style=\"border-width:0\" src=\"https://i.creativecommons.org/l/by/4.0/88x31.png\" /></a>"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}