GetSpecified.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# To Run Standard Searches\n",
    "- define new searches in mysets.py\n",
    "- beginning of notebook is assumed to be interactive until the requests have been checked\n",
    "- all progress and exception logging is done only for main loop"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `gsurl` is the GCS url for the dataset\n",
    "- `ds_dir` is our dataset identifier ( activity_id/institution_id/.../variable_id/grid_label ) NO version included"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import gcsfs\n",
    "import xarray as xr\n",
    "from datetime import datetime\n",
    "import itertools\n",
    "import os\n",
    "import gspread\n",
    "from oauth2client.service_account import ServiceAccountCredentials"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Local modules"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import myconfig\n",
    "from mysets import all_search\n",
    "from mydataset import dir2url_, dir2dict\n",
    "from mytasks import Check, Download, ReadFiles, SaveAsZarr, Upload, Cleanup\n",
    "from mysearch import esgf_search\n",
    "#from myidentify import gsurl2search, gsurl2dsdir\n",
    "from myutilities import search_df, remove_from_GC_bucket"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Directly connect to Search Status Page for updating\n",
    "# Anyone can view: https://docs.google.com/spreadsheets/d/1yAt7604tVt7OXXZUyL2uALtGP2WVa-Pb5NMuTluFsAc/edit?usp=sharing\n",
    "\n",
    "json_keyfile = '/home/naomi/json/CMIP6-d0cb1df722d1.json'  \n",
    "scope = ['https://spreadsheets.google.com/feeds','https://www.googleapis.com/auth/drive']\n",
    "credentials = ServiceAccountCredentials.from_json_keyfile_name(json_keyfile, scope)\n",
    "gc = gspread.authorize(credentials)\n",
    "sheet_name = \"CMIP6_UpdateSchedule\"\n",
    "sh = gc.open(sheet_name)\n",
    "wks = sh.worksheet(\"Searches\")\n",
    "\n",
    "# read the first row to get the column labels\n",
    "columns = wks.row_values(3)\n",
    "col_status = columns.index('status')\n",
    "col_run = columns.index('last run')\n",
    "col_drive = columns.index('HD')\n",
    "col_dataset = columns.index('current dataset')\n",
    "col_node = columns.index('search node')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# CONFIGURE ESGF Search here\n",
    "node_pref = myconfig.node_pref\n",
    "dtype = myconfig.dtype\n",
    "hd = '/d3'\n",
    "myconfig.local_target_prefix = hd + '/naomi/zarr-minimal/'\n",
    "dir2local = dir2url_(myconfig.local_target_prefix)\n",
    "\n",
    "clear_logs = True   # re-run a search from scratch - deleting prior logs\n",
    "update_ESGF = True\n",
    "update_Needed  = True\n",
    "\n",
    "search = 'D2d-1l'   # if search='test', must define below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row in Google Sheet: 22\n",
      "zarrs will be written to:  /d3/naomi/zarr-minimal/\n"
     ]
    }
   ],
   "source": [
    "# Update the Google Sheet Request form:\n",
    "\n",
    "try:\n",
    "    search_row = wks.find(search).row\n",
    "    print('Row in Google Sheet:',search_row)\n",
    "except:\n",
    "    print('need another row in Google Form!')\n",
    "    wks.append_row([search])\n",
    "    search_row = wks.find(search).row\n",
    "\n",
    "search_node = 'llnl'\n",
    "ESGF_site = dtype[search_node]\n",
    "\n",
    "date = datetime.now().strftime('%Y-%m-%d')\n",
    "wks.update_cell(search_row, col_run + 1, date)\n",
    "wks.update_cell(search_row, col_drive + 1, hd)\n",
    "wks.update_cell(search_row, col_node + 1, search_node)\n",
    "\n",
    "print('zarrs will be written to: ',myconfig.local_target_prefix)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# CONFIGURE GCS\n",
    "fs     = gcsfs.GCSFileSystem(token='anon', access='read_only',cache_timeout=-1)\n",
    "df_GCS = pd.read_csv('https://cmip6.storage.googleapis.com/pangeo-cmip6-noQC.csv', dtype='unicode')\n",
    "df_GCS['ds_dir'] = df_GCS.apply(lambda row: myconfig.target_format % row,axis=1)\n",
    "\n",
    "# make available to all modules\n",
    "myconfig.fs = fs\n",
    "myconfig.df_GCS = df_GCS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('D2d-1l',\n",
       " {'table_id': ['day'],\n",
       "  'experiment_id': ['1pctCO2',\n",
       "   'abrupt-4xCO2',\n",
       "   'historical',\n",
       "   'piControl',\n",
       "   'ssp119',\n",
       "   'ssp126',\n",
       "   'ssp245',\n",
       "   'ssp370',\n",
       "   'ssp434',\n",
       "   'ssp460',\n",
       "   'ssp534-over',\n",
       "   'ssp585'],\n",
       "  'variable_id': ['mrro', 'mrso', 'mrsos']})"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "if search in all_search:\n",
    "    # Use a pre-defined search\n",
    "    asearch = all_search[search]\n",
    "else:\n",
    "    # Define your own search: Pick keyword values to specify your search, not specifying a particular keyword means it will find all.\n",
    "    asearch = {\n",
    "                     'table_id'      : ['SImon']\n",
    "                    #,'activity_id'   : ['CDRMIP']\n",
    "                    ,'experiment_id' : ['historical']\n",
    "                    ,'variable_id'   : [\"siconc\"] \n",
    "                    #,'member_id'     : ['r2i1p1f1']\n",
    "                    #,'source_id'     : ['BCC-ESM1']\n",
    "                    ,'grid_label'    : ['gn']\n",
    "                }\n",
    "\n",
    "# check if ANOTHER notebook is doing the same search\n",
    "lock_file = f'logs/{search}.lock'\n",
    "if os.path.exists(lock_file):\n",
    "    if clear_logs:  \n",
    "        command = f'/bin/rm logs/*{search}*'\n",
    "        print(command)\n",
    "        os.system(command)\n",
    "    #else:\n",
    "    #    assert False, f'lockfile {lock_file} exists'\n",
    "        \n",
    "f = open(lock_file,'w')\n",
    "date = str(datetime.now().strftime(\"%Y%m%d-%H%M\"))\n",
    "f.write(f'started {search} at {date}')\n",
    "f.close()\n",
    "\n",
    "search, asearch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "ename": "FileNotFoundError",
     "evalue": "[Errno 2] File csv/ESGF_D2d-1l.csv does not exist: 'csv/ESGF_D2d-1l.csv'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mFileNotFoundError\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-7-9b3b353357e5>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     16\u001b[0m     \u001b[0mdf_ESGF\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mto_csv\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34mf'csv/ESGF_{search}.csv'\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0mindex\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mFalse\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     17\u001b[0m \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 18\u001b[0;31m     \u001b[0mdf_ESGF\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mpd\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mread_csv\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34mf'csv/ESGF_{search}.csv'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdtype\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'unicode'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     19\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     20\u001b[0m \u001b[0mlen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdf_ESGF\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mlen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdf_ESGF\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mds_dir\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0munique\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/local/python/anaconda3/envs/pangeo-Jul2020/lib/python3.7/site-packages/pandas/io/parsers.py\u001b[0m in \u001b[0;36mparser_f\u001b[0;34m(filepath_or_buffer, sep, delimiter, header, names, index_col, usecols, squeeze, prefix, mangle_dupe_cols, dtype, engine, converters, true_values, false_values, skipinitialspace, skiprows, skipfooter, nrows, na_values, keep_default_na, na_filter, verbose, skip_blank_lines, parse_dates, infer_datetime_format, keep_date_col, date_parser, dayfirst, cache_dates, iterator, chunksize, compression, thousands, decimal, lineterminator, quotechar, quoting, doublequote, escapechar, comment, encoding, dialect, error_bad_lines, warn_bad_lines, delim_whitespace, low_memory, memory_map, float_precision)\u001b[0m\n\u001b[1;32m    674\u001b[0m         )\n\u001b[1;32m    675\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 676\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0m_read\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfilepath_or_buffer\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mkwds\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    677\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    678\u001b[0m     \u001b[0mparser_f\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mname\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/local/python/anaconda3/envs/pangeo-Jul2020/lib/python3.7/site-packages/pandas/io/parsers.py\u001b[0m in \u001b[0;36m_read\u001b[0;34m(filepath_or_buffer, kwds)\u001b[0m\n\u001b[1;32m    446\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    447\u001b[0m     \u001b[0;31m# Create the parser.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 448\u001b[0;31m     \u001b[0mparser\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mTextFileReader\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfp_or_buf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwds\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    449\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    450\u001b[0m     \u001b[0;32mif\u001b[0m \u001b[0mchunksize\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0miterator\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/local/python/anaconda3/envs/pangeo-Jul2020/lib/python3.7/site-packages/pandas/io/parsers.py\u001b[0m in \u001b[0;36m__init__\u001b[0;34m(self, f, engine, **kwds)\u001b[0m\n\u001b[1;32m    878\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0moptions\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m\"has_index_names\"\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mkwds\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m\"has_index_names\"\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    879\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 880\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_make_engine\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mengine\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    881\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    882\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mclose\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/local/python/anaconda3/envs/pangeo-Jul2020/lib/python3.7/site-packages/pandas/io/parsers.py\u001b[0m in \u001b[0;36m_make_engine\u001b[0;34m(self, engine)\u001b[0m\n\u001b[1;32m   1112\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_make_engine\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mengine\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m\"c\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1113\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mengine\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;34m\"c\"\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1114\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_engine\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mCParserWrapper\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0moptions\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1115\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1116\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mengine\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;34m\"python\"\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/local/python/anaconda3/envs/pangeo-Jul2020/lib/python3.7/site-packages/pandas/io/parsers.py\u001b[0m in \u001b[0;36m__init__\u001b[0;34m(self, src, **kwds)\u001b[0m\n\u001b[1;32m   1889\u001b[0m         \u001b[0mkwds\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m\"usecols\"\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0musecols\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1890\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1891\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_reader\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mparsers\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mTextReader\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msrc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwds\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1892\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0munnamed_cols\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_reader\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0munnamed_cols\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1893\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32mpandas/_libs/parsers.pyx\u001b[0m in \u001b[0;36mpandas._libs.parsers.TextReader.__cinit__\u001b[0;34m()\u001b[0m\n",
      "\u001b[0;32mpandas/_libs/parsers.pyx\u001b[0m in \u001b[0;36mpandas._libs.parsers.TextReader._setup_parser_source\u001b[0;34m()\u001b[0m\n",
      "\u001b[0;31mFileNotFoundError\u001b[0m: [Errno 2] File csv/ESGF_D2d-1l.csv does not exist: 'csv/ESGF_D2d-1l.csv'"
     ]
    }
   ],
   "source": [
    "# make dataframe of ESGF search results\n",
    "\n",
    "if update_ESGF:\n",
    "    x = [value for key,value in asearch.items()]\n",
    "    searches = [p for p in itertools.product(*x)]\n",
    "\n",
    "    dESGF = []\n",
    "    for s in searches:\n",
    "        dsearch = dict(zip(asearch.keys(),s))\n",
    "        print(dsearch)\n",
    "        df = esgf_search(dsearch, server=ESGF_site)\n",
    "        if len(df)>0:\n",
    "            dESGF += [df]\n",
    "\n",
    "    df_ESGF = pd.concat(dESGF)\n",
    "    df_ESGF.to_csv(f'csv/ESGF_{search}.csv',index=False)\n",
    "else:\n",
    "    df_ESGF = pd.read_csv(f'csv/ESGF_{search}.csv', dtype='unicode')\n",
    "\n",
    "len(df_ESGF), len(df_ESGF.ds_dir.unique())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# make dataframe of all needed\n",
    "\n",
    "if update_Needed:\n",
    "    df_ESGF['cversion'] = [int(s[1:]) for s in df_ESGF.version_id]\n",
    "    df_ESGF = df_ESGF.sort_values(by=['cversion'])\n",
    "    df_ESGF = df_ESGF.drop_duplicates(subset =[\"ds_dir\",\"ncfile\"],keep='last')\n",
    " \n",
    "    #df_GCS['ds_dir'] = df_GCS.apply(lambda row: myconfig.target_format % row,axis=1) # THIS WON'T WORK FOR DCPP\n",
    "    df_GCS = df_GCS[df_GCS.version != 'unknown']\n",
    "    df_GCS['ds_dir']=[s.split('gs://cmip6/CMIP6/')[-1].split('/v20')[0] for s in df_GCS.zstore]\n",
    "    df_GCS['cversion'] = [int(s) for s in df_GCS.version]\n",
    "    df_GCS = df_GCS.sort_values(by=['version'])\n",
    "    df_GCS = df_GCS.drop_duplicates(subset =[\"ds_dir\"],keep='last')\n",
    "\n",
    "    df = pd.merge(df_ESGF, df_GCS, how='outer', on=['ds_dir'], indicator=True, suffixes=('', '_y'),)\n",
    "    df_new_dataset = df[df._merge == 'left_only'] \n",
    "    \n",
    "    df_check = df[df._merge == 'both'] # we might want to add these if the ESGF version is newer than the GCS version\n",
    "    # New versions (at least 2 days newer) which exist at ESGF\n",
    "    if len(df_check) > 0:\n",
    "        df_check['dversion'] = df_check.apply(lambda row: row.cversion - row.cversion_y,axis=1)\n",
    "        df_check = df_check.sort_values(by=['dversion'])\n",
    "        df_new_version = df_check[df_check.dversion > 2] # at least 2 days newer\n",
    "\n",
    "        # Want to append together df_new_dataset and df_new_version\n",
    "        keys = ['activity_id', 'institution_id', 'source_id', 'experiment_id', 'member_id', 'table_id', 'variable_id', 'grid_label', 'version_id', 'ncfile', 'file_size', 'url', 'data_node', 'ds_dir', 'node_order', 'start', 'stop']\n",
    "        df_new_version = df_new_version[keys]\n",
    "        df_new_dataset = df_new_dataset[keys]\n",
    "\n",
    "        df_needed = df_new_dataset.append(df_new_version)\n",
    "    else:\n",
    "        df_needed = df_new_dataset\n",
    "\n",
    "    df_needed['version'] = [s[1:] for s in df_needed.version_id]\n",
    "\n",
    "    num_stores = 0\n",
    "    if len(df_needed) > 0:\n",
    "        num_stores = df_needed.ds_dir.nunique() \n",
    "        print(f'needed: nfiles={len(df_needed)}, nstores={num_stores}')\n",
    "    \n",
    "        df_needed['member'] = [int(s.split('r')[-1].split('i')[0]) for s in df_needed['member_id']]\n",
    "        df_needed = df_needed.sort_values(by=['member'])\n",
    "        #df_needed['zsize'] = [df_needed[df_needed.ds_dir==zs]['file_size'].sum() for zs in df_needed['ds_dir']]\n",
    "        #df_needed = df_needed.sort_values(by=['zsize'])\n",
    "        df_needed.to_csv(f'csv/needed_{search}.csv',index=False)\n",
    "else:\n",
    "    df_needed = pd.read_csv(f'csv/needed_{search}.csv', dtype='unicode')\n",
    "\n",
    "print('Variables')\n",
    "try:\n",
    "    for var in df_needed.variable_id.unique():\n",
    "        print(var,df_needed[df_needed.variable_id==var].ds_dir.nunique())\n",
    "\n",
    "    print('\\nExperiments')\n",
    "    for exp in df_needed.experiment_id.unique():\n",
    "        print(exp,df_needed[df_needed.experiment_id==exp].ds_dir.nunique())\n",
    "except:\n",
    "    print('no new data available')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# make available to all modules\n",
    "myconfig.df_needed = df_needed\n",
    "ds_dirs = df_needed.ds_dir.unique()\n",
    "numdsets = len(ds_dirs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "progress_log  = f'logs/progress_{search}.log'\n",
    "failure_log   = f'logs/failure_{search}.log'\n",
    "success_log   = f'logs/success_{search}.log'\n",
    "logs = {1:progress_log, 2:failure_log, 3:success_log}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ierr:\n",
    "- 0 : proceed with next task\n",
    "- 1 : write to progress_log, go to next dataset (finished or try again)\n",
    "- 2 : write to failure_log,  go to next dataset (mark as un-usable - do not try again until problem is solved) \n",
    "- 3 : write to success_log,  go to next dataset (dataset added to cloud)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def write_log(file,str,verbose=True):\n",
    "    date = datetime.now().strftime('%Y%m%d-%H%M')\n",
    "    f = open(file,'a')\n",
    "    if verbose:\n",
    "        print(str)\n",
    "    f.write(f'{date}:{str}\\n')\n",
    "    f.close()\n",
    "    return"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# refresh the gcsfs\n",
    "fs.invalidate_cache()\n",
    "date = str(datetime.now().strftime(\"%Y%m%d-%H%M\"))\n",
    "for log in [1,2,3]:   \n",
    "    write_log(logs[log],f'starting at {date}') \n",
    "\n",
    "for item, ds_dir in enumerate(ds_dirs):\n",
    "    print(f'\\n{item}/{numdsets-1}',ds_dir)\n",
    "\n",
    "    #skip = 'EC-Earth3'\n",
    "    skip = 'none'\n",
    "    if skip in ds_dir:\n",
    "        write_log(progress_log,f'{ds_dir} skipping {skip}') \n",
    "        continue\n",
    "\n",
    "    version = df_needed[df_needed.ds_dir==ds_dir].version.values[0]\n",
    "\n",
    "    (ierr, exc) = Check(ds_dir, version, dir2local)\n",
    "    if ierr > 0:\n",
    "        write_log(logs[ierr],f'{ds_dir} {exc}'); continue\n",
    "                     \n",
    "    (gfiles, version, ierr, exc) = Download(ds_dir)\n",
    "    if ierr > 0:\n",
    "        write_log(logs[ierr],f'{ds_dir}, {exc}'); continue\n",
    "        \n",
    "    search_row = wks.find(search).row\n",
    "    wks.update_cell(search_row, col_dataset + 1, ds_dir)\n",
    "    \n",
    "    (ds,ierr,exc) = ReadFiles(ds_dir, gfiles, version, dir2dict)\n",
    "    \n",
    "    if ierr > 0:\n",
    "        write_log(logs[ierr],f'{ds_dir}, {exc}'); continue\n",
    "\n",
    "    (version,ierr,exc) = SaveAsZarr(ds_dir, ds, dir2local)\n",
    "    if ierr > 0:\n",
    "        write_log(logs[ierr],f'{ds_dir}, {exc}'); continue\n",
    "        \n",
    "    (zbdir, gsurl, ierr,exc) = Upload(ds_dir, version, dir2local)\n",
    "    if ierr > 0:\n",
    "        write_log(logs[ierr],f'{ds_dir}, {exc}'); continue\n",
    "\n",
    "    (ierr,exc) = Cleanup(ds_dir, version, gfiles, dir2local)\n",
    "    if ierr > 0:\n",
    "        write_log(logs[ierr],f'{ds_dir}, {exc}'); continue\n",
    "\n",
    "    date = datetime.now().strftime('%H:%M, %b%d')\n",
    "    status_str = f'{item+1} of {numdsets} at {date}'\n",
    "    search_row = wks.find(search).row\n",
    "    wks.update_cell(search_row, col_status + 1, status_str)\n",
    "\n",
    "    write_log(success_log,f'{zbdir} saved to {gsurl}'); continue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "pangeo-Jul2020",
   "language": "python",
   "name": "pangeo-jul2020"
  },
  "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}