Deduplication_SCCASC.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# This script reads metadata in a .csv, cleans and formats the data, and exports the result as a .csv for use in climate research."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from os.path import isfile\n",
    "from os import stat, listdir\n",
    "from datetime import datetime\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fix version attribute"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_version(row):\n",
    "    if row.version[0] != \"v\":\n",
    "        attributes = row.local_file.split(\"/\")\n",
    "        return attributes[-2]  # version is included as second to last item in local_file\n",
    "    return row.version\n",
    "\n",
    "df.version = df.apply(extract_version, axis=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Correct file paths"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def update_path(row):\n",
    "    path = row['local_file']\n",
    "    if not isfile(path):\n",
    "        if \"/data2/synda/sdt/data\" in path:\n",
    "            path = path.replace(\"/sdt\", \"\")\n",
    "        elif \"/data4/gsmwork/data/\" in path:\n",
    "            path = path.replace(\"gsmwork\", \"synda\")\n",
    "            if not isfile(path):\n",
    "                path = path.replace(\"/output/\", \"/output/output/\")\n",
    "                path = path.replace(\"/output1/\", \"/output1/output1/\")\n",
    "                path = path.replace(\"/output2/\", \"/output2/output2/\")\n",
    "    if not isfile(path):\n",
    "        raise Exception(f\"Could not update path for: {row['local_file']}\")\n",
    "    return path\n",
    "\n",
    "df['SCCASC_climatedata_path'] = df.apply(update_path, axis=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Confirm all files exist on filesystem"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df['file_exists'] = df['SCCASC_climatedata_path'].apply(lambda x: isfile(x))\n",
    "assert len(df[~df['file_exists']]) == 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Add OSCER Schooner paths"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def newpath(item):\n",
    "    root = item.split(\"/\")[1]\n",
    "    remaining_path = \"/\".join(item.split(\"/\")[2:])\n",
    "    if root == \"data\":\n",
    "        fullpath = f\"/condo/climatedata3/{remaining_path}\"\n",
    "    else:\n",
    "        fullpath = f\"/condo/climate{root}/{remaining_path}\"\n",
    "    return fullpath\n",
    "\n",
    "df['OSCER_schooner_path'] = df['SCCASC_climatedata_path'].apply(newpath)\n",
    "df['local_file'] = df['SCCASC_climatedata_path']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Segment out unique and duplicated entries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "unique = df[~df.duplicated(['filename'], False)]\n",
    "duplicated = df[df.duplicated(['filename'], False)]\n",
    "print(f\"{len(unique)} unique files\")\n",
    "print(f\"{len(duplicated)} duplicated files\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deduplicate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mtime = lambda x: datetime.fromtimestamp(stat(x).st_mtime).isoformat()\n",
    "issues = []\n",
    "def max_version(grp):\n",
    "    \"\"\" return single record from group \"\"\"\n",
    "    \n",
    "    # max version based on string comparison\n",
    "    records = grp[grp.version == grp.version.max()]\n",
    "    \n",
    "    if len(records) > 1:\n",
    "        # most recent modified timestamp\n",
    "        records['file_mtime'] = records.apply(lambda x: mtime(x.local_file), axis=1)\n",
    "        records = records[records.file_mtime == records.file_mtime.max()]\n",
    "    \n",
    "    if len(records) > 1:\n",
    "        # item with the most files in the same directory\n",
    "        keep = None\n",
    "        max_count = 0\n",
    "        for record in records.to_dict(orient='records'):\n",
    "            path, file = os.path.split(record['local_file'])\n",
    "            file_count = len(listdir(path))\n",
    "            if file_count > max_count:\n",
    "                keep = record\n",
    "                max_count = file_count\n",
    "        records = pd.DataFrame([keep])\n",
    "    \n",
    "    if not len(records) == 1:\n",
    "        issues.append(records.local_file)\n",
    "    \n",
    "    if issues:\n",
    "        raise Exception(\"Could not determine which record to use for the following: \", issues)\n",
    "    \n",
    "    return records\n",
    "    \n",
    "deduplicated = duplicated.groupby(['filename'], as_index=False).apply(max_version)\n",
    "print(f\"{len(deduplicated)} deduplicated files\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Merge unique and deduplicated"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Merging unique and deduplicated...\")\n",
    "cleaned = pd.concat([unique, deduplicated], ignore_index=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Remove records that are after year 2101"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cleaned['beg_year'] = cleaned.time.str.split(\"-\").apply(lambda x: int(x[0][0:4]))\n",
    "cleaned = cleaned[cleaned.beg_year < 2101]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sort"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cleaned = cleaned.sort_values(['variable', 'model', 'experiment', 'time_frequency', 'filename'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Add human readable variable names and dimensions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_variable_names = pd.read_csv('variable_name_lookup_table.csv')\n",
    "df_variable_dimensions = pd.read_csv('variable_dimension_lookup_table.csv')\n",
    "df_tmp = pd.merge(cleaned, df_variable_names, on='variable')\n",
    "df_complete = pd.merge(df_tmp, df_variable_dimensions, on='variable')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Export selected columns to CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "headers = (\n",
    "    'variable', 'variable_standard_name', 'variable_long_name', 'institute',\n",
    "    'model', 'domain', 'dimensions', 'project', 'realm', 'ensemble',\n",
    "    'experiment', 'time_frequency', 'time', 'version', 'filename', 'size',\n",
    "    'OSCER_schooner_path', 'SCCASC_climatedata_path', 'checksum_type', 'checksum',\n",
    ")\n",
    "df_complete.to_csv(\"CMIP5_climatedata_full.csv\", index=False, columns=headers)"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}