era5_in_gee_functions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src='./img/logos.png' alt='Logo EU Copernicus EUMETSAT' align='right' width='15%'></img>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ERA5 in GEE - Functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook lists all `functions` that are useful to bring ERA5 reanalysis data into Google Earth Engine.\n",
    "\n",
    "The functions can be grouped into the following categories:\n",
    "\n",
    "**[Useful data handling functions](#useful_functions)**\n",
    "* [createFileList](#create_filelist)\n",
    "* [createListOfLists](#create_list_of_lists)\n",
    "* [getEpochTimes](#epoch_times)\n",
    "* [getEpochTimes_daily](#epoch_times_daily)\n",
    "* [getEpochTimes_monthly](#epoch_times_monthly)\n",
    "\n",
    "**[Functions to generate a GeoTiff file with gdal](#generate_geotiff)**\n",
    "* [initTiff](#initTiff)\n",
    "* [createTiff](#create_tiff)\n",
    "* [getScaleFactor](#scale_factor)\n",
    "* [getOffset](#offset)\n",
    "* [setSpatialReference](#spatial_ref)\n",
    "\n",
    "**[Functions to convert NetCDF files to GeoTiffs](#convert_ncs_to_geotiffs)**\n",
    "* [ncToTiff](#nc_tiff)\n",
    "* [ncToTiff_hourly](#nc_tiff_hourly)\n",
    "* [convertFilesToTiff](#convert_to_tiff)\n",
    "\n",
    "**[Functions to temporally aggregate data](#aggregate)**\n",
    "* [createDailyFiles](#aggregate_daily)\n",
    "* [createMonthlyFiles](#aggregate_monthly)\n",
    "\n",
    "**[Functions to create / update manifests](#manifests)**\n",
    "* [updateManifest_hourly](#manifest_hourly)\n",
    "* [updateManifest_daily](#manifest_daily)\n",
    "* [updateManifest_monthly](#manifest_monthly)\n",
    "* [manifestToJSON](#manifest_json)\n",
    "* [createManifestCombined_hourly](#manifest_combined_hourly)\n",
    "* [createManifestCombined_daily](#manifest_combined_daily)\n",
    "* [createManifestCombined_monthly](#manifest_combined_monthly)\n",
    "\n",
    "**[Functions to upload files to Google Cloud Platform](#gcp_upload)**\n",
    "* [upload_blob](#upload_blob)\n",
    "* [uploadMonthlyFileToGCP](#upload_gcp_monthly)\n",
    "* [uploadToGCP](#upload_gcp)\n",
    "\n",
    "**[Command to ingest files on GCP into Earht Engine with manifest upload](#ee_manifest_upload)**\n",
    "* [ee_ingest](#ee_ingest)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Load libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from datetime import datetime, timedelta\n",
    "from dateutil.relativedelta import relativedelta\n",
    "import glob\n",
    "from osgeo import gdal, osr\n",
    "import pytz\n",
    "import re\n",
    "import json\n",
    "from google.cloud import storage\n",
    "import xarray as xr"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='useful_functions'></a>Useful data handling functions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='create_filelist'></a>`createFileList`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createFileList(directory,file_pattern):\n",
    "    ''' Creates a list of files based on a given file pattern\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to the file directory\n",
    "    file_pattern (str): File pattern of files to be included in the list\n",
    "    \n",
    "    Returns:\n",
    "    list: List of files\n",
    "    '''\n",
    "    os.chdir(directory)\n",
    "    return glob.glob(file_pattern)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='create_list_of_lists'></a>`createListOfLists`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createListOfLists(directory_list,aggregation,year):\n",
    "    ''' Creates a list of lists to create manifests with multiple variables\n",
    "    \n",
    "    Parameters:\n",
    "    directory_list (list): List of directory paths to tiff files\n",
    "    aggregation (str): string indicating the aggregation level, e.g. daily to be appended to the directory paths\n",
    "    year (str): year for which the list is created\n",
    "    \n",
    "    Returns:\n",
    "    fileList: List of tiff file lists (all parameters that shall be part of one EE asset)\n",
    "    '''\n",
    "    fileList=[]\n",
    "    for i in directory_list:\n",
    "        os.chdir(i)\n",
    "        # Create a file list for each entry of the directory list\n",
    "        fileList_tmp = createFileList(i,'./tiff/'+aggregation+'/'+year+'/*')\n",
    "        # Sort the resulting file list\n",
    "        fileList_tmp.sort()\n",
    "        # Append to build up a list of lists\n",
    "        fileList.append(fileList_tmp)\n",
    "        os.chdir('..')\n",
    "    return(fileList)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='epoch_times'></a>`getEpochTimes`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getEpochTimes(file, noOfBands):\n",
    "    ''' Converts the time information of a NetCDF file with 24 hourly time stamps from the Climate Data Store into \n",
    "    a list of epoch time stamps, which are required to ingest an asset to Earth Engine.\n",
    "    \n",
    "    Parameters:\n",
    "    file (netCDF4 Dataset): netCDF4 Dataset object\n",
    "    noOfBands (int): number of time stamps of the NetCDF Dataset\n",
    "    \n",
    "    Returns:\n",
    "    ls_epochtime (list): list of converted epoch time stamps\n",
    "    '''   \n",
    "    base = datetime(1900,1,1,0,0,0,0).replace(tzinfo=pytz.UTC)\n",
    "    ls_epochtime = []\n",
    "    \n",
    "    for i in range(1,noOfBands+1):\n",
    "        tmp = file.GetRasterBand(i)\n",
    "        tmp_time = tmp.GetMetadata()['NETCDF_DIM_time']\n",
    "        epoch_time = base + timedelta(hours=int(tmp_time))\n",
    "        ls_epochtime.append(int(epoch_time.timestamp()))\n",
    "    epoch_time = base + timedelta(hours=int(tmp_time)+1)\n",
    "    ls_epochtime.append(int(epoch_time.timestamp()))\n",
    "    return ls_epochtime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='epoch_times_daily'></a>`getEpochTimes_daily`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getEpochTimes_daily(year,month,day):\n",
    "    ''' Converts the time information given with year, month and day to the equivalent epoch time stamp.\n",
    "    \n",
    "    Parameters:\n",
    "    year (int): year\n",
    "    month (int): month\n",
    "    day (int): day\n",
    "    \n",
    "    Returns:\n",
    "    ls_epochtime (list): Converted epoch time stamps for start and end time of the asset\n",
    "    ''' \n",
    "    ls_epochtime = []\n",
    "    startTime = datetime(year,month,day, tzinfo=pytz.utc)\n",
    "    endTime = startTime + timedelta(days=1)\n",
    "    ls_epochtime.append(startTime.timestamp())\n",
    "    ls_epochtime.append(endTime.timestamp())\n",
    "    return ls_epochtime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='epoch_times_monthly'></a>`getEpochTimes_monthly`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getEpochTimes_monthly(year,month):\n",
    "    ''' Converts the time information given with year and month to the equivalent epoch time stamp.\n",
    "    \n",
    "    Parameters:\n",
    "    year (int): year\n",
    "    month (int): month\n",
    "    \n",
    "    Returns:\n",
    "    ls_epochtime (list): Converted epoch time stamps for start and end time of the asset\n",
    "    ''' \n",
    "    ls_epochtime = []\n",
    "    startTime = datetime(year,month, 1, tzinfo=pytz.utc)\n",
    "    endTime = startTime + relativedelta(months=+1)\n",
    "    ls_epochtime.append(startTime.timestamp())\n",
    "    ls_epochtime.append(endTime.timestamp())\n",
    "    return ls_epochtime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='generate_geotiff'></a>Functions to generate a GeoTiff with `gdal`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='init_tiff'></a>`initTiff`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def initTiff(filename, file, noOfBands):\n",
    "    ''' Initializes a tiff file based on a given NetCDF file and a geotransform object with 0.25 deg / 0.25 deg resolution.\n",
    "    \n",
    "    Parameters:\n",
    "    filename (str): name of the resulting GeoTiff file\n",
    "    file (NetCDF object): NetCDF object open with gdal.Open(file)\n",
    "    noOfBands (int): number of bands of the resulting GeoTiff\n",
    "    \n",
    "    Returns:\n",
    "    outFile (gdal TIFF object): returns a Tiff file object that can be used to write array information with func(createTiff)\n",
    "    ''' \n",
    "    outFile = gdal.GetDriverByName('GTiff').Create(filename, file.RasterXSize, file.RasterYSize, noOfBands, gdal.GDT_Float32)\n",
    "    geotransform = (-180.0, 0.25, 0.0, 90.0, 0.0, -0.25)\n",
    "    outFile.SetGeoTransform(geotransform)\n",
    "    return outFile"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='create_tiff'></a>`createTiff`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createTiff(file, outfile, scale_factor, offset):\n",
    "    ''' Writes array information (raster bands) from a NetCDF file to a Tiff object which was initialized with func(initTiff).\n",
    "    \n",
    "    Parameters:\n",
    "    file (NetCDF file object): NetCDF file object opened with gdal.Open()\n",
    "    outFile (GeoTiff object): GeoTiff object initialized with func(initTiff)\n",
    "    scale_factor: scale factor of the NetCDF file retrieved with func(getScaleFactor)\n",
    "    offset: offset value of the NetCDF file retrieved with func(getOffset)\n",
    "    \n",
    "    Returns:\n",
    "    outBand (gdal TIFF object): returns a Tiff file object that can be saved with .FlushCashe()\n",
    "    ''' \n",
    "    for j in range(1, file.RasterCount+1):\n",
    "        fileLayer = file.GetRasterBand(j).ReadAsArray().astype('float')\n",
    "        finalArray = float(offset) + (fileLayer * float(scale_factor))\n",
    "        finalArray[finalArray<0] = 0.0\n",
    "        outBand = outfile.GetRasterBand(j)\n",
    "        outBand.WriteArray(finalArray)\n",
    "    return outBand"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='scale_factor'></a>`getScaleFactor`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getScaleFactor(file, parameter):\n",
    "    ''' Returns the scale factor from a NetCDF file as float\n",
    "    \n",
    "    Parameters:\n",
    "    file (NetCDF file object): NetCDF file object opened with gdal.Open()\n",
    "    parameter (str): Specify the parameter of the data values\n",
    "    \n",
    "    Returns:\n",
    "    scale factor as float\n",
    "    '''\n",
    "    return float(file.GetMetadataItem(parameter+\"#scale_factor\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='offset'></a>`getOffset`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getOffset(file, parameter):\n",
    "    ''' Returns the offset from a NetCDF file as float\n",
    "    \n",
    "    Parameters:\n",
    "    file (NetCDF file object): NetCDF file object opened with gdal.Open()\n",
    "    parameter (str): Specify the parameter of the data values\n",
    "    \n",
    "    Returns:\n",
    "    offset as float\n",
    "    '''\n",
    "    return float(file.GetMetadataItem(parameter+\"#add_offset\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='spatial_ref'></a>`setSpatialReference`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def setSpatialReference(file,EPSGCode):\n",
    "    ''' Sets the spatial reference to a GeoTiff object initiated with func(initTiff)\n",
    "    \n",
    "    Parameters:\n",
    "    file (GeoTiff object): GeoTiff object initiated with func(initTiff)\n",
    "    EPSGCode(int): epsg code of the resulting spatial reference\n",
    "    '''\n",
    "    # Initiate a SpatialReference object\n",
    "    srs = osr.SpatialReference()\n",
    "    # Retrieve the spatial reference information from an epsg code\n",
    "    srs.ImportFromEPSG(EPSGCode)\n",
    "    # Set the spatial reference object to the GeoTiff file\n",
    "    file.SetProjection(srs.ExportToWkt())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='convert_ncs_to_geotiffs'></a>Functions to convert NetCDF files to GeoTiffs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='nc_tiff'></a>`ncToTiff`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ncToTiff(file, noOfBands, epsgCode,outfile):\n",
    "    ''' Function that combines various steps to convert an aggregated NetCDF file (daily or monthly) to a GeoTiff file. Scale and Offset factors do not\n",
    "    need to be applied, as those were already accounted for while the data was aggregated with xarray. See funct(createDailyFiles)\n",
    "    or func(createMonthlyFiles).\n",
    "    \n",
    "    Parameters:\n",
    "    file (str): Path to a NetCDF file\n",
    "    noOfBands (int): number of bands of the resulting GeoTiff file\n",
    "    epsgCode(int): epsc code number\n",
    "    outfile(str): Name of resulting GeoTiff file\n",
    "    '''\n",
    "    # Open a NetCDF file\n",
    "    ncFile=gdal.Open(file)\n",
    "    # Initiate a GeoTiff object\n",
    "    outTiff = initTiff(outfile,ncFile,noOfBands)\n",
    "    \n",
    "    fileLayer = ncFile.GetRasterBand(1).ReadAsArray().astype('float')\n",
    "    fileLayer[fileLayer<0] = 0.0\n",
    "    outBand = outTiff.GetRasterBand(1)\n",
    "    outBand.WriteArray(fileLayer)\n",
    "    # Set spatial reference to the GeoTiff object\n",
    "    setSpatialReference(outTiff, epsgCode)\n",
    "    # Write the GeoTiff file and close it\n",
    "    outBand.FlushCache()\n",
    "    outTiff=None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='nc_tiff_hourly'></a>`ncToTiff_hourly`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ncToTiff_hourly(file, noOfBands, epsgCode, outfile, parameter):\n",
    "    ''' Function that combines various steps to convert a NetCDF file with 24 hourly time steps to a GeoTiff file\n",
    "    with 24 bands. Scale and Offset factors are applied during the conversion.\n",
    "    \n",
    "    Parameters:\n",
    "    file (str): Path to a NetCDF file\n",
    "    noOfBands (int): number of bands of the resulting GeoTiff file\n",
    "    epsgCode(int): epsc code number\n",
    "    outfile(str): Name of resulting GeoTiff file\n",
    "    parameter(str): name of the parameter in the NetCDF file\n",
    "    '''\n",
    "    if(parameter=='maximum_2m_temperature_since_previous_post_processing'):\n",
    "        parameter='mx2t'\n",
    "    elif(parameter=='minimum_2m_temperature_since_previous_post_processing'):\n",
    "        parameter='mn2t'\n",
    "    elif(parameter=='surface_pressure'):\n",
    "        parameter='sp'\n",
    "    elif(parameter=='2m_dewpoint_temperature'):\n",
    "        parameter='d2m'\n",
    "    elif(parameter=='mean_sea_level_pressure'):\n",
    "        parameter='msl'\n",
    "    elif(parameter=='10m_u_component_of_wind'):\n",
    "        parameter='u10'\n",
    "    elif(parameter=='10m_v_component_of_wind'):\n",
    "        parameter='v10'\n",
    "    elif(parameter=='t2m'):\n",
    "        parameter='t2m'\n",
    "    else:\n",
    "        parameter='tp'\n",
    "    ncFile = gdal.Open(file)\n",
    "    outTiff = initTiff(outfile,ncFile,noOfBands)\n",
    "    scale_factor = getScaleFactor(ncFile, parameter)\n",
    "    offset = getOffset(ncFile, parameter)\n",
    "\n",
    "    outBand = createTiff(ncFile, outTiff, scale_factor, offset)\n",
    "    setSpatialReference(outTiff,epsgCode)\n",
    "    outBand.FlushCache()\n",
    "    outTiff=None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='convert_to_tiff'></a>`convertFilesToTiff`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convertFilesToTiff(directory, time_step, parameter, year, epsg):\n",
    "    ''' Function that loops through a directory with NetCDF files and converts the files to GeoTiff files. Calls the\n",
    "    functions 'ncToTiff' or 'ncToTiff_hourly'.\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to directory with NetCDF files\n",
    "    time_step (str): additon to the directory path differentiating between different aggregation levels\n",
    "    parameter(str): additon to the directory path to specify the parameter\n",
    "    year(str): additional to the directory path to differentiate the year\n",
    "    epsg(str): epsg code of the resulting GeoTiff file\n",
    "    '''\n",
    "    fileList = createFileList(directory, './era5_'+parameter+'/nc/'+time_step+'/'+year+'/era5_surface_pressure_1985_06_12*.nc')\n",
    "    # Test if a file of one year is missing\n",
    "    if(len(fileList)<365):\n",
    "        return\n",
    "\n",
    "    fileList.sort()\n",
    "    for file in fileList:\n",
    "        tmp = file.split('/')\n",
    "        print(tmp[5][:-3])\n",
    "        # Generate name of the outfile\n",
    "        outfile = directory+'era5_'+parameter+'/tiff/'+time_step+'/'+year+'/'+str(tmp[5][:-3])+'.tif'\n",
    "        print(outfile)\n",
    "        if(time_step!='hourly'):\n",
    "            # if daily or monthly files are converted, use func(ncToTiff) else func(ncToTiff_hourly)\n",
    "            ncToTiff(file,1,year,epsg, outfile)\n",
    "        else:\n",
    "            ncToTiff_hourly(file,24,year, epsg,outfile,parameter)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='aggregate'></a>Functions to temporally aggregate data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='aggregate_daily'></a>`createDailyFiles`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createDailyFiles(directory, parameter, year, aggregation):\n",
    "    ''' Function that loops over a list of daily NetCDF files with 24 time stamps and aggregates (resamples) the files\n",
    "    to the daily mean, sum, min or maximum. For precipitation, two NetCDF files are loaded, a ERA5 Total precipitation \n",
    "    is a forecast parameter. This means that the precipitation of the 00 time stamp is the accumulation of the rain fallen\n",
    "    between 23 and 00. Thus, we need to retrieve the data of the first time step of the following file.\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to directory with NetCDF files\n",
    "    parameter (str): parameter to be resampled. If 'tp', aggregation will be based on two NetCDF files\n",
    "    year(str): addition to the directory path\n",
    "    aggregation(str): what type of aggregation shall be executed - mean, min, max, sum\n",
    "    '''    \n",
    "    fileList = createFileList(directory, './era5_'+parameter+'/nc/hourly/'+year+'/era5_'+parameter+'_'+year+'*')\n",
    "    fileList.sort()\n",
    "    \n",
    "    for i in range(0,len(fileList)-1):\n",
    "        # if paramter if total precipitation, open two subsequent NetCDF files and concat the files on the time dimension\n",
    "        if(parameter=='tp'):\n",
    "            array=xr.open_mfdataset([fileList[i],fileList[i+1]],concat_dim='time', combine='nested')\n",
    "            print(array)\n",
    "        else:\n",
    "        # else, open the NetCDF file and apply automatically scale and offset factors by setting the kwarg mask_and_scale=True\n",
    "            array = xr.open_dataset(fileList[i], mask_and_scale=True, decode_times=True)\n",
    "            print(array)\n",
    "        tmp = fileList[i].split('/')\n",
    "\n",
    "        # Define the name of the aggregated NetCDF file\n",
    "        outFileName = directory+'./era5_'+parameter+'/nc/daily/'+year+'/'+tmp[5][:-3]+'_daily_'+aggregation+'.nc'\n",
    "        \n",
    "        print(outFileName)\n",
    "        \n",
    "        # Offer different aggregation methods and aggregate on a daily basis\n",
    "        if(aggregation=='mean'):\n",
    "            print('mean')\n",
    "            array.resample(time='1D').mean().to_netcdf(outFileName, mode='w', compute=True)\n",
    "        # Total precipitation values are summed over one day. By setting the keyword argument \"closed='right'\", xarray\n",
    "        # automatically drops the first time step of the first day and takes the first time of the next day\n",
    "        elif(aggregation=='sum'):\n",
    "            print('sum')\n",
    "            array.resample(time='1D',closed='right').sum().isel(time=1).to_netcdf(outFileName, mode='w', compute=True)\n",
    "        elif(aggregation=='min'):\n",
    "            print('min')\n",
    "            array.resample(time='1D').min().to_netcdf(outFileName, mode='w', compute=True)\n",
    "        else:\n",
    "            print('max')\n",
    "            array.resample(time='1D').max().to_netcdf(outFileName, mode='w', compute=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='aggregate_monthly'></a>`createMonthlyFiles`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createMonthlyFiles(directory, parameter, year, aggregation):\n",
    "    ''' Function that loops over a list of daily NetCDF files with 24 time stamps and aggregates (resamples) the files\n",
    "    to the monthly mean, sum, min or maximum.\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to directory with NetCDF files\n",
    "    parameter (str): parameter to be resampled. If 'tp', aggregation will be based on two NetCDF files\n",
    "    year(str): addition to the directory path\n",
    "    aggregation(str): what type of aggregation shall be executed - mean, min, max, sum\n",
    "    ''' \n",
    "    month_list = ['01','02','03','04','05','06','07','08','09','10','11','12']\n",
    "\n",
    "    for i in month_list:\n",
    "        fileList_param = createFileList(directory,'./era5_'+parameter+'/nc/'+year+'/era5_'+parameter+'_'+year+'_'+i+'*')\n",
    "        fileList_param.sort()\n",
    "\n",
    "        os.chdir(directory)\n",
    "        array_param = xr.open_mfdataset(fileList_param,combine='nested', concat_dim='time')\n",
    "\n",
    "        tmp = fileList_param[0].split('/')\n",
    "        outFileName_param = directory+'./era5_'+parameter+'/nc/monthly/'+year+'/'+tmp[4][:-6]+'_monthly_'+aggregation+'.nc'\n",
    "\n",
    "        # Account for different aggregation levels\n",
    "        if(aggregation=='mean'):\n",
    "            print('mean')\n",
    "            array_param.resample(time='1M').mean().to_netcdf(outFileName_param, mode='w', compute=True)\n",
    "        elif(aggregation=='sum'):\n",
    "            print('sum')\n",
    "            array_param.resample(time='1M').sum().to_netcdf(outFileName_param, mode='w', compute=True)\n",
    "        elif(aggregation=='min'):\n",
    "            print('min')\n",
    "            array_param.resample(time='1M').min().to_netcdf(outFileName_param, mode='w', compute=True)\n",
    "        else:\n",
    "            print('max')\n",
    "            array_param.resample(time='1M').max().to_netcdf(outFileName_param, mode='w', compute=True)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='manifests'></a>Functions to create / update manifests"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_hourly'></a>`updateManifest_hourly`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def updateManifest_hourly(directory, eeCollectionName, assetName, startTime, endTime, bandIndex, gs_bucket_list, uris1, uris2, uris3, uris4, uris5, uris6, uris7, uris8, uris9, year,month, day, hour):\n",
    "    ''' Function that opens an example manifest structure file for ERA5 hourly assets and updates the dictionary items\n",
    "    accordingly.\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to directory with NetCDF files\n",
    "    eeCollectionName(str):  Path to collection name on Earth Engine\n",
    "    assetName(str): name of resulting asset in Earth Engine\n",
    "    startTime(int): start time in epoch time\n",
    "    endTime(int): end time in epoch time\n",
    "    bandIndex(int): number of band\n",
    "    gs_bucket_list: list of GCP buckets holding the tiff files that shall be part of the asset\n",
    "    uris1-uris9 (str): name of various tiff files uploaded to GCP\n",
    "    year(str): add as additional asset information - year\n",
    "    month(str): add as additional asset information - month\n",
    "    day(str): add as additional asset information - year\n",
    "    hour(str): add as additional asset information - year\n",
    "    \n",
    "    Returns:\n",
    "    jsonFile object\n",
    "    ''' \n",
    "    with open(directory+'manifest_structure_hourly.json','r') as f:\n",
    "        jsonFile = json.load(f)\n",
    "\n",
    "    jsonFile['name']=eeCollectionName+assetName\n",
    "    jsonFile['tilesets'][0]['sources'][0]['uris']='gs://'+gs_bucket_list[0]+'/'+uris1\n",
    "    jsonFile['tilesets'][1]['sources'][0]['uris']='gs://'+gs_bucket_list[1]+'/'+uris2\n",
    "    jsonFile['tilesets'][2]['sources'][0]['uris']='gs://'+gs_bucket_list[2]+'/'+uris3\n",
    "    jsonFile['tilesets'][3]['sources'][0]['uris']='gs://'+gs_bucket_list[3]+'/'+uris4\n",
    "    jsonFile['tilesets'][4]['sources'][0]['uris']='gs://'+gs_bucket_list[4]+'/'+uris5\n",
    "    jsonFile['tilesets'][5]['sources'][0]['uris']='gs://'+gs_bucket_list[5]+'/'+uris6\n",
    "    jsonFile['tilesets'][6]['sources'][0]['uris']='gs://'+gs_bucket_list[6]+'/'+uris7\n",
    "    jsonFile['tilesets'][7]['sources'][0]['uris']='gs://'+gs_bucket_list[7]+'/'+uris8\n",
    "    jsonFile['tilesets'][8]['sources'][0]['uris']='gs://'+gs_bucket_list[8]+'/'+uris9\n",
    "\n",
    "    jsonFile['bands'][0]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][1]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][2]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][3]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][4]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][5]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][6]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][7]['tileset_band_index']=bandIndex\n",
    "    jsonFile['bands'][8]['tileset_band_index']=bandIndex\n",
    "\n",
    "    jsonFile['start_time']['seconds']=startTime\n",
    "    jsonFile['end_time']['seconds']=endTime\n",
    "    jsonFile['properties']['year']=year\n",
    "    jsonFile['properties']['month']=month\n",
    "    jsonFile['properties']['day']=day \n",
    "    jsonFile['properties']['hour']=hour\n",
    "    return jsonFile"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_daily'></a>`updateManifest_daily`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def updateManifest_daily(directory, eeCollectionName, assetName, startTime, endTime, gs_bucket_list, uris1, uris2, uris3, uris4, uris5, uris6, uris7, uris8, uris9, year,month, day):\n",
    "    ''' Function that opens an example manifest structure file for ERA5 daily assets and updates the dictionary items\n",
    "    accordingly.\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to directory with NetCDF files\n",
    "    eeCollectionName(str):  Path to collection name on Earth Engine\n",
    "    assetName(str): name of resulting asset in Earth Engine\n",
    "    startTime(int): start time in epoch time\n",
    "    endTime(int): end time in epoch time\n",
    "    gs_bucket_list: list of GCP buckets holding the tiff files that shall be part of the asset\n",
    "    uris1-uris9 (str): name of various tiff files uploaded to GCP\n",
    "    year(str): add as additional asset information - year\n",
    "    month(str): add as additional asset information - month\n",
    "    day(str): add as additional asset information - year\n",
    "    \n",
    "    Returns:\n",
    "    jsonFile object\n",
    "    ''' \n",
    "    with open(directory+'manifest_structure_daily.json','r') as f:\n",
    "        jsonFile = json.load(f)\n",
    "\n",
    "    jsonFile['name']=eeCollectionName+assetName\n",
    "    jsonFile['tilesets'][0]['sources'][0]['uris']='gs://'+gs_bucket_list[0]+'/'+uris1\n",
    "    jsonFile['tilesets'][1]['sources'][0]['uris']='gs://'+gs_bucket_list[1]+'/'+uris2\n",
    "    jsonFile['tilesets'][2]['sources'][0]['uris']='gs://'+gs_bucket_list[2]+'/'+uris3\n",
    "    jsonFile['tilesets'][3]['sources'][0]['uris']='gs://'+gs_bucket_list[3]+'/'+uris4\n",
    "    jsonFile['tilesets'][4]['sources'][0]['uris']='gs://'+gs_bucket_list[4]+'/'+uris5\n",
    "    jsonFile['tilesets'][5]['sources'][0]['uris']='gs://'+gs_bucket_list[5]+'/'+uris6\n",
    "    jsonFile['tilesets'][6]['sources'][0]['uris']='gs://'+gs_bucket_list[6]+'/'+uris7\n",
    "    jsonFile['tilesets'][7]['sources'][0]['uris']='gs://'+gs_bucket_list[7]+'/'+uris8\n",
    "    jsonFile['tilesets'][8]['sources'][0]['uris']='gs://'+gs_bucket_list[8]+'/'+uris9 \n",
    "    jsonFile['start_time']['seconds']=startTime\n",
    "    jsonFile['end_time']['seconds']=endTime\n",
    "    jsonFile['properties']['year']=year\n",
    "    jsonFile['properties']['month']=month\n",
    "    jsonFile['properties']['day']=day   \n",
    "    return jsonFile"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_monthly'></a>`updateManifest_monthly`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def updateManifest_monthly(directory,eeCollectionName, assetName, startTime, endTime, gs_bucket_list, uris1, uris2, uris3, uris4, uris5, uris6, uris7, uris8, uris9, year, month):\n",
    "    ''' Function that opens an example manifest structure file for ERA5 monthly assets and updates the dictionary items\n",
    "    accordingly.\n",
    "    \n",
    "    Parameters:\n",
    "    directory (str): Path to directory with NetCDF files\n",
    "    eeCollectionName(str):  Path to collection name on Earth Engine\n",
    "    assetName(str): name of resulting asset in Earth Engine\n",
    "    startTime(int): start time in epoch time\n",
    "    endTime(int): end time in epoch time\n",
    "    gs_bucket_list: list of GCP buckets holding the tiff files that shall be part of the asset\n",
    "    uris1-uris9 (str): name of various tiff files uploaded to GCP\n",
    "    year(str): add as additional asset information - year\n",
    "    month(str): add as additional asset information - month\n",
    "    \n",
    "    Returns:\n",
    "    jsonFile object\n",
    "    ''' \n",
    "    \n",
    "    with open(directory+'manifest_structure_monthly.json','r') as f:\n",
    "        jsonFile = json.load(f)\n",
    "\n",
    "    jsonFile['name']=eeCollectionName+assetName\n",
    "    jsonFile['tilesets'][0]['sources'][0]['uris']='gs://'+gs_bucket_list[0]+'/'+uris1\n",
    "    jsonFile['tilesets'][1]['sources'][0]['uris']='gs://'+gs_bucket_list[1]+'/'+uris2\n",
    "    jsonFile['tilesets'][2]['sources'][0]['uris']='gs://'+gs_bucket_list[2]+'/'+uris3\n",
    "    jsonFile['tilesets'][3]['sources'][0]['uris']='gs://'+gs_bucket_list[3]+'/'+uris4\n",
    "    jsonFile['tilesets'][4]['sources'][0]['uris']='gs://'+gs_bucket_list[4]+'/'+uris5\n",
    "    jsonFile['tilesets'][5]['sources'][0]['uris']='gs://'+gs_bucket_list[5]+'/'+uris6\n",
    "    jsonFile['tilesets'][6]['sources'][0]['uris']='gs://'+gs_bucket_list[6]+'/'+uris7\n",
    "    jsonFile['tilesets'][7]['sources'][0]['uris']='gs://'+gs_bucket_list[7]+'/'+uris8\n",
    "    jsonFile['tilesets'][8]['sources'][0]['uris']='gs://'+gs_bucket_list[8]+'/'+uris9\n",
    "    jsonFile['start_time']['seconds']=startTime\n",
    "    jsonFile['end_time']['seconds']=endTime\n",
    "    jsonFile['properties']['year']=year\n",
    "    jsonFile['properties']['month']=month\n",
    "    return jsonFile"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_json'></a>`manifestToJSON`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def manifestToJSON(manifestDict, path,outFile):\n",
    "    ''' Function that dumps a json file object and creates a JSON file\n",
    "    \n",
    "    Parameters:\n",
    "    manifestDict(json object):\n",
    "    path(str): path where JSON file shall be stored\n",
    "    outFile(str): name of the resulting JSON file\n",
    "    ''' \n",
    "    with open(path+outFile+'.json','w') as fp:\n",
    "        json.dump(manifestDict,fp,indent=4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_combined_hourly'></a>`createManifestCombined_hourly`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createManifestCombined_hourly(fileList, ncFileList, year, bucket_list, directory_manifest,directory_outfile):\n",
    "    ''' Function that loops over a fileList and creates manifest files for ERA5 hourly assets.\n",
    "    \n",
    "    Parameters:\n",
    "    fileList (list): List of file list of all variables that will be part of the EE asset\n",
    "    ncFileList (list): list of NetCDF files in order to retrieve the time steps and be able to convert them to epoch times\n",
    "    year(str): addition to outfile name\n",
    "    bucket_list(list): list of GCP buckets holding files to be ingested to Earth Engine\n",
    "    directory_manifest(str): path to example manifests\n",
    "    directory_outfile(str): path where manifest files shall be stored\n",
    "    ''' \n",
    "    for i in range(0,len(fileList[0])):\n",
    "        print(len(fileList[0]))\n",
    "        item = list(zip(*fileList))[i]\n",
    "\n",
    "        tmp = re.findall('\\d+', item[0])\n",
    "    \n",
    "        # Create assetName based on year month and day information\n",
    "        assetName=tmp[3]+tmp[4]+tmp[5]\n",
    "        \n",
    "        # open a NetCDF file in order to retrieve the time stamps\n",
    "        ncFile = gdal.Open(ncFileList[i])\n",
    "\n",
    "        # Convert the time stamps to epoch times\n",
    "        ls_epochtimes = getEpochTimes(ncFile,24)\n",
    "\n",
    "        uris_list = []\n",
    "        for i in item:\n",
    "             tmp2 = i.split('/')\n",
    "             uris_list.append(tmp2[4])\n",
    "        print(uris_list)\n",
    "\n",
    "        for k in range(0,len(ls_epochtimes)-1):\n",
    "            print(k)\n",
    "            hour= str(k).zfill(2)\n",
    "            # For all 24 epoch times, create manifest\n",
    "            manifest = updateManifest_hourly(directory=directory_manifest,\n",
    "                                  eeCollectionName='projects/earthengine-legacy/assets/projects/ecmwf/era5_hourly/',\n",
    "                                  assetName=assetName+'T'+hour,\n",
    "                                  startTime=int(ls_epochtimes[k]),\n",
    "                                  endTime=int(ls_epochtimes[k+1]),\n",
    "                                  bandIndex=k,\n",
    "                                  gs_bucket_list=bucket_list,\n",
    "                                  uris1=uris_list[0],\n",
    "                                  uris2=uris_list[1],\n",
    "                                  uris3=uris_list[2],\n",
    "                                  uris4=uris_list[3],\n",
    "                                  uris5=uris_list[4],\n",
    "                                  uris6=uris_list[5],\n",
    "                                  uris7=uris_list[6],\n",
    "                                  uris8=uris_list[7],\n",
    "                                  uris9=uris_list[8],\n",
    "                                  year=int(tmp[3]),\n",
    "                                  month=int(tmp[4]),\n",
    "                                  day=int(tmp[5]),\n",
    "                                  hour=int(hour))\n",
    "            outfile='manifest_'+assetName+hour+'_hourly'\n",
    "            # Save JSON object \n",
    "            manifestToJSON(manifest,directory_outfile+year+'/',outfile)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_combined_daily'></a>`createManifestCombined_daily`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createManifestCombined_daily(fileList, year,bucket_list, directory_manifest,directory_outfile):\n",
    "    ''' Function that loops over a fileList and creates manifest files for ERA5 daily assets.\n",
    "    \n",
    "    Parameters:\n",
    "    fileList (list): List of file list of all variables that will be part of the EE asset\n",
    "    year(str): addition to outfile name\n",
    "    bucket_list(list): list of GCP buckets holding files to be ingested to Earth Engine\n",
    "    directory_manifest(str): path to example manifests\n",
    "    directory_outfile(str): path where manifest files shall be stored\n",
    "    ''' \n",
    "    for i in range(0,len(fileList[0])):\n",
    "        item = list(zip(*fileList))[i]\n",
    "\n",
    "        tmp = re.findall('\\d+', item[0])\n",
    "        assetName=tmp[3]+tmp[4]+tmp[5]\n",
    "        \n",
    "        # Get start and end times of the asset in epoch times\n",
    "        ls_epochtimes = getEpochTimes_daily(int(tmp[3]),int(tmp[4]),int(tmp[5]))\n",
    "        \n",
    "        uris_list = []\n",
    "        for i in item:\n",
    "            tmp2 = i.split('/')\n",
    "            uris_list.append(tmp2[4])\n",
    "            \n",
    "        # Update manifest information\n",
    "        manifest = updateManifest_daily(directory=directory_manifest,\n",
    "                                        eeCollectionName='projects/earthengine-legacy/assets/projects/ecmwf/era5_daily/',\n",
    "                                        assetName=assetName,\n",
    "                                        startTime = int(ls_epochtimes[0]),\n",
    "                                        endTime = int(ls_epochtimes[1]),\n",
    "                                        gs_bucket_list = bucket_list,\n",
    "                                        uris1=uris_list[0],\n",
    "                                        uris2=uris_list[1],\n",
    "                                        uris3=uris_list[2],\n",
    "                                        uris4=uris_list[3],\n",
    "                                        uris5=uris_list[4],\n",
    "                                        uris6=uris_list[5],\n",
    "                                        uris7=uris_list[6],\n",
    "                                        uris8=uris_list[7],\n",
    "                                        year=int(tmp[3]),\n",
    "                                        month=int(tmp[4]),\n",
    "                                        day=int(tmp[5]))\n",
    "        outfile='manifest_'+assetName+'_daily'\n",
    "        # Save JSON object\n",
    "        manifestToJSON(manifest,directory_outfile+year+'/',outfile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='manifest_combined_monthly'></a>`createManifestCombined_monthly`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def createManifestCombined_monthly(fileList, year,bucket_list, directory_manifest,directory_outfile):\n",
    "    ''' Function that loops over a fileList and creates manifest files for ERA5 monthly assets.\n",
    "    \n",
    "    Parameters:\n",
    "    fileList (list): List of file list of all variables that will be part of the EE asset\n",
    "    year(str): addition to outfile name\n",
    "    bucket_list(list): list of GCP buckets holding files to be ingested to Earth Engine\n",
    "    directory_manifest(str): path to example manifests\n",
    "    directory_outfile(str): path where manifest files shall be stored\n",
    "    ''' \n",
    "    for i in range(0,len(fileList[0])):\n",
    "\n",
    "        item = list(zip(*fileList))[i]\n",
    "\n",
    "        tmp = re.findall('\\d+', item[0])\n",
    "        assetName=tmp[3]+tmp[4]\n",
    "        \n",
    "        # Get start and end times of the asset in epoch times        \n",
    "        ls_epochtimes = getEpochTimes_monthly(int(tmp[3]),int(tmp[4]))\n",
    "        \n",
    "        uris_list = []\n",
    "        for i in item:\n",
    "            tmp2 = i.split('/')\n",
    "            print(tmp2)\n",
    "            uris_list.append(tmp2[4])\n",
    "            \n",
    "        # Update manifest   \n",
    "        manifest = updateManifest_monthly(directory=directory_manifest,\n",
    "                                        eeCollectionName='projects/earthengine-legacy/assets/projects/ecmwf/era5_monthly/',\n",
    "                                        assetName=assetName,\n",
    "                                        startTime = int(ls_epochtimes[0]),\n",
    "                                        endTime = int(ls_epochtimes[1]),\n",
    "                                        gs_bucket_list = bucket_list,\n",
    "                                        uris1=uris_list[0],\n",
    "                                        uris2=uris_list[1],\n",
    "                                        uris3=uris_list[2],\n",
    "                                        uris4=uris_list[3],\n",
    "                                        uris5=uris_list[4],\n",
    "                                        uris6=uris_list[5],\n",
    "                                        uris7=uris_list[6],\n",
    "                                        uris8=uris_list[7],\n",
    "                                        year=int(tmp[3]),\n",
    "                                        month=int(tmp[4]))\n",
    "        outfile='manifest_'+assetName+'_monthly'\n",
    "        # Save JSON object\n",
    "        manifestToJSON(manifest,directory_outfile+year+'/',outfile)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='gcp_upload'></a>Functions to upload files to Google Cloud Platform (GCP)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='upload_blob'></a>`upload_blob`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def upload_blob(bucket_name, source_file_name, destination_blob_name):\n",
    "    ''' Function that uploads a file to Google Cloud Platform.\n",
    "    \n",
    "    Parameters:\n",
    "    bucket_name(str): name of bucket on GCP\n",
    "    source_file_name(str): name of local file to be uploaded\n",
    "    destination_blob_name(str): name of file on GCP\n",
    "    ''' \n",
    "    storage_client = storage.Client()\n",
    "    bucket = storage_client.get_bucket(bucket_name)\n",
    "    blob = bucket.blob(destination_blob_name)\n",
    "    if(blob.exists()):\n",
    "        print('File {} already exists'.format(destination_blob_name))\n",
    "        next\n",
    "    else:\n",
    "        blob.upload_from_filename(source_file_name)\n",
    "\n",
    "        print('File {} uploaded to {}.'.format(\n",
    "                source_file_name,\n",
    "                destination_blob_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='upload_gcp_monthly'></a>`uploadMonthlyFilesToGCP`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def uploadMonthlyFilesToGCP(directory,parameter,year,bucket):\n",
    "    ''' Function that uploads monthly files to Google Cloud Platform.\n",
    "    \n",
    "    Parameters:\n",
    "    directory(str): path to directory with files to be uploaded\n",
    "    parameter(str): parameter name - addition to source file name\n",
    "    year(str): year - addition to source file name\n",
    "    bucket(str): name of bucket on GCP\n",
    "    ''' \n",
    "    fileList = createFileList(directory,'./era5_'+parameter+'/tiff/monthly/'+year+'/*.tif')\n",
    "    fileList.sort()\n",
    "    for file in fileList:\n",
    "        tmp = file.split('/')\n",
    "        print(tmp)\n",
    "        destname = tmp[5]\n",
    "        print(destname)\n",
    "        upload_blob(bucket,file,destname)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='upload_gcp'></a>`uploadToGCP`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def uploadToGCP(directory,year,time_step,parameter,bucket):\n",
    "    ''' Function that uploads a file to Google Cloud Platform.\n",
    "    \n",
    "    Parameters:\n",
    "    directory(str): path to directory with files to be uploaded\n",
    "    year(str): year - addition to source file name\n",
    "    time_step(str): time step - addition to source file name\n",
    "    parameter(str): parameter name - addition to source file name\n",
    "    bucket(str): name of bucket on GCP\n",
    "    ''' \n",
    "    fileList = createFileList(directory, 'era5_'+parameter+'/tiff/'+time_step+'/'+year+'/*.tif')\n",
    "    fileList.sort()\n",
    "\n",
    "    for file in fileList:\n",
    "        print(file)\n",
    "        tmp = file.split('/')\n",
    "        print(tmp)\n",
    "\n",
    "        upload_blob(bucket,file,tmp[4])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## <a id='ee_manifest_upload'></a>Command to ingest files on GCP to Earth Engine with manifest upload"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <a id='ee_ingest'></a>`ee_ingest`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ee_ingest(manifest_list):\n",
    "    ''' Function that calls the earthengine Python API command to ingest files stored on GCP into Earth Engine\n",
    "    based on manifest upload.\n",
    "    \n",
    "    Parameters:\n",
    "    mainfest_list(list): path to manifests to upload\n",
    "    '''\n",
    "    for i in manifest_list:\n",
    "        print(i)\n",
    "        cmd = 'earthengine --use_cloud_api upload image --force --manifest ' + i\n",
    "        print(cmd)\n",
    "        os.system(cmd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a rel=\"license\" href=\"http://creativecommons.org/licenses/by-nc/4.0/\"><img alt=\"Creative Commons License\" style=\"border-width:0\" src=\"https://i.creativecommons.org/l/by-nc/4.0/88x31.png\" align='right' /></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.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}