[dev]replacing_preprocessed_stacks_lost_with_stor01.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In 2/8/2020, there was a loss of the stor01 storage, including the external storage for the stack.PreprocessedStack table.  As a result, there are still entries for the affected items in the table, but fetches for the stacks for those entries would fail.  When the database was migrated from dj 0.11 to dj 0.12.9, Anthony Ramos automatically detected the missing external files, and replaced them with a short text message:\n",
    "\n",
    "'External storage was lost on 2/8/2020, as a result the external for this stack is unavailable. \\n                    The message you are seeing now is a result of a migration to dj 0.12.9 on 8/25/2021. If you would like this stack \\n                    repopulated, please contact the pipeline engineer '\n",
    "\n",
    "In order to correct this loss, it is not feasible to simply repopulate the entries in stack.PreprocessedStack, because there are too many downstream dependencies that would also be deleted and repopulated as a result, including published results that we do not want to lose the original records.  \n",
    "\n",
    "Instead, we directly modify the files that are stored at the location indicated in the external path.\n",
    "\n",
    "THIS SHOULD NOT BE A ROUTINE OPERATION.  THIS IS AN UNUSUAL CASE.\n",
    "\n",
    "However, this allows us to replace those files, using the same code that was used to routinely populate them the first time, since we have prior knowledge that the code has not changed since that time.  This allows the population of additional tables that are dependent on those stacks, or using those stacks for other purposes (figures, export, etc).  \n",
    "\n",
    "The below function was written by Anthony Ramos and modified by Paul Fahey for overwriting the error message text files with the correct stack files.  It operates by tracking the storage location from the external file using pymysql, recreates the stacks using prepackaged functions from the stack pipeline, and writes them to those file locations.  \n",
    "\n",
    "Notably, dj 0.12.9 has a preliminary caching function that was elaborated in future versions (dj 0.13).  If you import stack, it sets dj.config['cache'] to /tmp/dj-cache.  The stack code will also preferentially fetch from the local cache if there is a dj.config['cache'] key.  This is usually preferable, since it will speed up the fetch, and it is very rare and violates fundamental data storage principles to modify the entries in the table without deleting and repopulating.  As mentioned above, this is a niche case that should not be generalized.  However, in this case, if you fetch the error message, that fetch will go into the cache.  Therefore, even if you modify the stack at the external location, fetchs to that entry will still return the error message from the cache instead of the stack, even though at the remote location that file has been successfully overwritten.  This can be avoided by simply deleting the dj.config['cache'] item, as seen in the import block below.  \n",
    "\n",
    "All of the preprocessed stack_candidate_keys below have been replaced using this notebook, as of 2/18/2022."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Loading local settings from pipeline_config.json\n",
      "Connecting pfahey@at-database.ad.bcm.edu:3306\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import scipy\n",
    "import pymysql\n",
    "import numpy as np\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "import datajoint as dj\n",
    "from pathlib import Path\n",
    "from datajoint.utils import safe_write\n",
    "from datajoint.blob import pack, unpack\n",
    "from pipeline.utils import registration, enhancement\n",
    "from pipeline.stack import CorrectedStack, PreprocessedStack\n",
    "\n",
    "# necessary to prevent datajoint from automatically caching the external fetchs, \n",
    "# preventing the updated files from showing.  cache automatically set during stack import, maybe other places\n",
    "del dj.config['cache']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'animal_id': '21067', 'session': '9', 'stack_idx': '1'},\n",
       " {'animal_id': '21067', 'session': '10', 'stack_idx': '25'},\n",
       " {'animal_id': '21067', 'session': '13', 'stack_idx': '2'},\n",
       " {'animal_id': '22620', 'session': '4', 'stack_idx': '16'},\n",
       " {'animal_id': '22620', 'session': '4', 'stack_idx': '16'},\n",
       " {'animal_id': '22620', 'session': '5', 'stack_idx': '14'},\n",
       " {'animal_id': '22846', 'session': '2', 'stack_idx': '20'},\n",
       " {'animal_id': '22846', 'session': '2', 'stack_idx': '20'},\n",
       " {'animal_id': '22846', 'session': '7', 'stack_idx': '14'},\n",
       " {'animal_id': '22846', 'session': '10', 'stack_idx': '17'},\n",
       " {'animal_id': '23343', 'session': '5', 'stack_idx': '22'},\n",
       " {'animal_id': '23555', 'session': '5', 'stack_idx': '13'},\n",
       " {'animal_id': '23656', 'session': '14', 'stack_idx': '18'},\n",
       " {'animal_id': '23964', 'session': '4', 'stack_idx': '23'},\n",
       " {'animal_id': '26644', 'session': '14', 'stack_idx': '18'},\n",
       " {'animal_id': '26645', 'session': '2', 'stack_idx': '19'}]"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dics = np.load('/mnt/scratch07/zhiwei/static_scan_release_keys_filtered.npy', allow_pickle=True)\n",
    "stack_candidate_keys = [{k:v for k,v in zip(('animal_id','session','stack_idx'),\n",
    "                                            d['stack'].split('-'))} for d in dics]\n",
    "stack_candidate_keys"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'animal_id': 21067,\n",
       " 'session': 10,\n",
       " 'stack_idx': 25,\n",
       " 'volume_id': 1,\n",
       " 'channel': 1}"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "key = (PreprocessedStack & stack_candidate_keys[1]).fetch1('KEY')\n",
    "display(key)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(array(['External storage was lost on 2/8/2020, as a result the external for this stack is unavailable. \\n                    The message you are seeing now is a result of a migration to dj 0.12.9 on 8/25/2021. If you would like this stack \\n                    repopulated, please contact the pipeline engineer '],\n",
       "       dtype=object),\n",
       " array(['External storage was lost on 2/8/2020, as a result the external for this stack is unavailable. \\n                    The message you are seeing now is a result of a migration to dj 0.12.9 on 8/25/2021. If you would like this stack \\n                    repopulated, please contact the pipeline engineer '],\n",
       "       dtype=object),\n",
       " array(['External storage was lost on 2/8/2020, as a result the external for this stack is unavailable. \\n                    The message you are seeing now is a result of a migration to dj 0.12.9 on 8/25/2021. If you would like this stack \\n                    repopulated, please contact the pipeline engineer '],\n",
       "       dtype=object))"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# if the stack has not been replaced yet, will return a text file describing the cause of the loss\n",
    "# if replaced, will return stack matrices\n",
    "r,l,s = (PreprocessedStack & key).fetch1('resized','lcned','sharpened')\n",
    "r,l,s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# if the stack has not been replaced yet, will fail\n",
    "# if replaced, will plot a slice from the preprocessed stacks\n",
    "fig,axes = plt.subplots(1,3,figsize=(10,3))\n",
    "for ax,s in zip(axes,(r,l,s)):\n",
    "    ax.imshow(s[100,:,:])\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# if the stacks have not been replaced yet, this function will recompute the 3 preprocessed stacks\n",
    "# from corrected stack, look up the 3 external locations where the stack is supposed to be, and \n",
    "# overwrite the 3 files at those locations with the recomputed stacks\n",
    "\n",
    "\n",
    "# replace_stack(key)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def replace_stack(key):\n",
    "    connection = pymysql.connect(host='at-database.ad.bcm.edu',\n",
    "                             user='root',\n",
    "                             password='primate123',\n",
    "                             database='pipeline_stack',\n",
    "                             cursorclass=pymysql.cursors.DictCursor)\n",
    "    with connection.cursor() as cursor:\n",
    "        ## Gets all the hashes/filepaths for the given stack key\n",
    "        cursor.execute(f\"\"\"SELECT * FROM pipeline_stack.`~external_stack` e \n",
    "            JOIN pipeline_stack.__preprocessed_stack p \n",
    "                ON e.hash = p.resized \n",
    "                    OR e.hash = p.lcned \n",
    "                    OR e.hash = p.sharpened \n",
    "            WHERE filepath is not null AND \n",
    "                p.animal_id={key['animal_id']} \n",
    "                AND p.stack_idx={key['stack_idx']} \n",
    "                AND p.session={key['session']}\n",
    "                AND p.volume_id={key['volume_id']}\n",
    "                AND p.channel={key['channel']}\"\"\")\n",
    "        \n",
    "        stacks = cursor.fetchall()\n",
    "        \n",
    "        #find_paths has the logic to determine which of the results from the join corresponds to which\n",
    "        # stack. Since it's a join this is needed\n",
    "        paths = find_paths(stacks)\n",
    "        \n",
    "        resized_hash = paths['resized']['filepath']\n",
    "        lcned_hash = paths['lcned']['filepath']\n",
    "        sharpened_hash = paths['sharpened']['filepath']\n",
    "        \n",
    "        #### START PIPELINE CODE \n",
    "        # Load stack\n",
    "        folder = dj.config['stores']['stack']['location']\n",
    "\n",
    "        stack = (CorrectedStack() & key).get_stack(key['channel'])\n",
    "\n",
    "        # Resize to be 1 um^3\n",
    "\n",
    "        um_sizes = (CorrectedStack & key).fetch1('um_depth', 'um_height', 'um_width')\n",
    "        resized = registration.resize(stack, um_sizes, desired_res=1)\n",
    "\n",
    "        # Enhance\n",
    "        lcned = enhancement.lcn(resized, (3, 25, 25))\n",
    "\n",
    "\n",
    "        # Sharpen\n",
    "        sharpened = enhancement.sharpen_2pimage(lcned, 1)\n",
    "        \n",
    "        ### END PIPELINE CODE\n",
    "        \n",
    "        # need to pack the arrays the same way datajoint does so on a dj fetch it returns the right result\n",
    "        resized_blob = pack(resized)\n",
    "        lcned_blob = pack(lcned)\n",
    "        sharpened_blob = pack(sharpened)\n",
    "\n",
    "        \n",
    "        \n",
    "        # building all the paths\n",
    "        resized_path = os.path.join(folder,resized_hash)\n",
    "        lcned_path = os.path.join(folder,lcned_hash)\n",
    "        sharpened_path = os.path.join(folder,sharpened_hash)\n",
    "        \n",
    "        # bypassing the safe_write allows it to overwrite the existing error message file\n",
    "        print(resized_path)\n",
    "        Path(resized_path).write_bytes(resized_blob)\n",
    "        # safe_write(resized_path,resized_blob)\n",
    "        print(lcned_path)\n",
    "        Path(lcned_path).write_bytes(lcned_blob)\n",
    "        # safe_write(lcned_path,lcned_blob)\n",
    "        print(sharpened_path)\n",
    "        Path(sharpened_path).write_bytes(sharpened_blob)\n",
    "        # safe_write(sharpened_path,sharpened_blob)\n",
    "\n",
    "\n",
    "def find_paths(results):\n",
    "    paths = {}\n",
    "    for stack in results:\n",
    "        \n",
    "        if(stack['hash'] == stack['resized']):\n",
    "            paths['resized'] = stack\n",
    "        elif (stack['hash'] == stack['lcned']):\n",
    "            paths['lcned'] = stack\n",
    "        elif (stack['hash'] == stack['sharpened']) :\n",
    "            paths['sharpened'] = stack\n",
    "        \n",
    "    return paths\n",
    "    \n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}