qc-pass-fail-aggregation.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#setup dependencies, install metric parser\n",
    "import sys\n",
    "!{sys.executable} -m pip install --no-cache-dir --upgrade  qc-metric-aggregator\n",
    "\n",
    "import os\n",
    "import os.path\n",
    "import firecloud.api as fapi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#set up constants\n",
    "bucket = os.environ['WORKSPACE_BUCKET']\n",
    "workspace_namespace = os.environ['WORKSPACE_NAMESPACE']\n",
    "workspace_name = os.environ['WORKSPACE_NAME']\n",
    "threshold_file_name = \"thresholds.yml\"\n",
    "final_output_file_name = \"qc_results.tsv\"\n",
    "error_output_file_name = 'samples_with_errors'\n",
    "\n",
    "#we should pull this from a central place rather than a workspace specific bucket\n",
    "master_thresholds_file = bucket + \"/\" + threshold_file_name\n",
    "#copy the thresholds file to the notebook env\n",
    "!gsutil cp $master_thresholds_file ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": []
   },
   "outputs": [],
   "source": [
    "#fetch sample ids from the terra workspace table\n",
    "samples = fapi.get_entities(workspace_namespace, workspace_name, \"sample\").json()\n",
    "sample_ids = [s['name'] for s in samples]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": []
   },
   "outputs": [],
   "source": [
    "#figure out which files from cromwell runs we will need to localize this heuristic can definitely be futher optimized\n",
    "#allow for resuming a failed workflow by checking if the file already exists\n",
    "if not os.path.isfile('files_to_localize'):\n",
    "    !gsutil ls -r $bucket/** | grep -v -E 'stderr|stdout|\\.sh$|\\.log$|/pipelines-logs/output|/rc$|/script$|\\.pdf$' > files_to_localize\n",
    "else:\n",
    "    print(\"Found existing file list.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#prepare the working dir and read in already processed samples\n",
    "!mkdir -p localized_files\n",
    "completed_samples = {}\n",
    "if not os.path.isfile(final_output_file_name):\n",
    "    print(\"No previous results, processing all samples.\")\n",
    "else:\n",
    "    line_number = 1\n",
    "    with open(final_output_file_name, 'r') as existing_file:\n",
    "        for line in existing_file:\n",
    "            if line_number >= 2:\n",
    "                sample_name = line.split(\"\\t\")[0]\n",
    "                completed_samples[sample_name] = True\n",
    "            line_number += 1\n",
    "            \n",
    "    print(f'Loaded {len(completed_samples)} completed samples')            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#run the metric aggregator for each sample and write out the results\n",
    "from process_metrics.threshold_file_parser import ThresholdFileParser\n",
    "from process_metrics.qc_validator import QcValidator\n",
    "from process_metrics.metrics import AvailableMetrics\n",
    "from process_metrics.report_generator import ReportGenerator\n",
    "\n",
    "pass_fail_thresholds = ThresholdFileParser(threshold_file_name).thresholds()\n",
    "\n",
    "qc_results = []    \n",
    "samples_with_errors = []\n",
    "\n",
    "if len(completed_samples) == 0: \n",
    "    first_sample = True\n",
    "else:\n",
    "    first_sample = False\n",
    "\n",
    "#if we're resuming, the cursor will be placed at the end of the final line, so we need to print a newline\n",
    "append_newline = os.path.isfile(final_output_file_name) and os.path.getsize(final_output_file_name) > 0\n",
    "\n",
    "with open(final_output_file_name, 'a') as fout:  \n",
    "    if append_newline:\n",
    "        fout.write(\"\\n\")\n",
    "    for sample_id in sample_ids:\n",
    "        #don't reprocess a sample that we've already processed\n",
    "        if sample_id in completed_samples:\n",
    "            #print(f'Already Processed {sample_id}.')\n",
    "            continue\n",
    "        try:\n",
    "            #localize files relevant to this sample\n",
    "            _ = !cat files_to_localize | grep $sample_id | grep -v .cram | gsutil -m cp -I ./localized_files\n",
    "        \n",
    "            metrics = AvailableMetrics(sample_id)\n",
    "            validator = QcValidator(\"localized_files/\")\n",
    "            res = ReportGenerator(sample_id, pass_fail_thresholds, metrics, validator).gather_metrics()\n",
    "            headers = res[0]\n",
    "            values = res[1]\n",
    "            if first_sample:\n",
    "                first_sample = False\n",
    "                headers[0] = \"entity:qc_result_sample_id\"\n",
    "                lowercased_headers = [h.lower() for h in headers]\n",
    "                print(str.join(\"\\t\", lowercased_headers), file=fout)\n",
    "            print(f'Writing row for {sample_id}')\n",
    "            print(str.join(\"\\t\", values), file=fout)\n",
    "        except:\n",
    "            print(f'ERROR with {sample_id}')\n",
    "            samples_with_errors.append(sample_id)\n",
    "        finally:\n",
    "            #clean up localized files for this sample\n",
    "            !rm -rf localized_files/*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#write out samples with errors\n",
    "with open(error_output_file_name, 'w') as fout:\n",
    "    print(str.join(\"\\n\", samples_with_errors), file=fout)\n",
    "\n",
    "#copy samples with errors to the workspace bucket\n",
    "uploaded_errors = f'{bucket}/{error_output_file_name}'\n",
    "!gsutil cp $error_output_file_name $uploaded_errors\n",
    "    \n",
    "#copy the results into terra as a datatable\n",
    "fapi.upload_entities_tsv(workspace_namespace, workspace_name, final_output_file_name, \"flexible\")\n",
    "\n",
    "#copy the TSV to the workspace bucket\n",
    "uploaded_tsv = bucket + '/' + final_output_file_name\n",
    "!gsutil cp $final_output_file_name $uploaded_tsv"
   ]
  }
 ],
 "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.8.5"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}