notebooks/1.RDATtoDataFrame.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e6e9c093",
   "metadata": {},
   "source": [
    "# Generate Dataframe from RDAT\n",
    "\n",
    "This notebook takes an input RDAT file with reactivity data for a given sequence library,\n",
    "and extracts the relevant data and stores it in a Dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c1a3501b-a96a-4529-9a29-8418c6868955",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.append('..')\n",
    "\n",
    "import os\n",
    "import pandas as pd\n",
    "from openknotscore.utils import load_rdat, get_global_blank_out\n",
    "\n",
    "# Trick to display dataframes with scrollbars\n",
    "from IPython.display import display, HTML\n",
    "display(HTML(\"<style>.jp-OutputArea-output {display:flex}</style>\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9e7ec2e5-d281-488c-9d92-999de6fe7c99",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate empty dataframe with columns that we will extract from the RDAT\n",
    "\n",
    "df = pd.DataFrame(columns=[\"id\",\"title\",\"author\",\"sequence\",\"reads\",\"signal_to_noise\",\"snr\",\"warning\",\"reactivity\",\"errors\"])\n",
    "\n",
    "for file in os.listdir('../data/rdats'):\n",
    "    if not file.endswith('.rdat'):\n",
    "        continue\n",
    "    rdat, constructName = load_rdat(f'../data/rdats/{file}')\n",
    "\n",
    "    seqList = []\n",
    "\n",
    "    BLANK_OUT5, BLANK_OUT3 = get_global_blank_out(rdat.constructs[constructName])\n",
    "\n",
    "    # Loop through all sequences in the RDAT, extract relevant data, and add to the dataframe\n",
    "    for sequence in rdat.constructs[constructName].data:\n",
    "        # Grab annotations\n",
    "        id = None\n",
    "        author = None\n",
    "        title = None\n",
    "        score_start_idx = None\n",
    "        score_end_idx = None\n",
    "        if 'Eterna' in sequence.annotations:\n",
    "            for annot in sequence.annotations.get('Eterna'):\n",
    "                if annot.startswith('id:'):\n",
    "                    id = annot.removeprefix('id:')\n",
    "                if annot.startswith('author:'):\n",
    "                    author = annot.removeprefix('author:')\n",
    "                if annot.startswith('design_name:'):\n",
    "                    title = annot.removeprefix('design_name:')\n",
    "                if annot.startswith('score:start_idx:'):\n",
    "                    score_start_idx = int(annot.removeprefix('score:start_idx:'))\n",
    "                if annot.startswith('score:end_idx:'):\n",
    "                    score_end_idx = int(annot.removeprefix('score:end_idx:'))\n",
    "        if title is None:\n",
    "            title = sequence.annotations.get('name')[0]\n",
    "        if id is None:\n",
    "            id = f'EXTERNAL:{title}'\n",
    "        seq = sequence.annotations.get('sequence')[0]\n",
    "        reads = None\n",
    "        if 'reads' in sequence.annotations:\n",
    "            reads = int(sequence.annotations.get('reads')[0])\n",
    "        signal_to_noise = sequence.annotations.get('signal_to_noise')[0]\n",
    "        snr = float(sequence.annotations.get('signal_to_noise')[0].split(\":\")[1])\n",
    "        warning = sequence.annotations.get('warning', '-')[0]\n",
    "        if score_start_idx is None:\n",
    "            score_start_idx = BLANK_OUT5 + 1\n",
    "        if score_end_idx is None:\n",
    "            score_end_idx = len(seq) - BLANK_OUT3\n",
    "\n",
    "        # Get reactivity data and errors\n",
    "        reactivity = [float('nan')]*BLANK_OUT5 + sequence.values + [float('nan')]*BLANK_OUT3\n",
    "        errors = [float('nan')]*BLANK_OUT5 + sequence.errors + [float('nan')]*BLANK_OUT3\n",
    "\n",
    "        # Create a dataframe from this row\n",
    "        row = pd.DataFrame(\n",
    "            data = [[id,seq,reads,signal_to_noise,snr,warning,reactivity,errors,score_start_idx,score_end_idx]],\n",
    "            columns=[\"id\",\"sequence\",\"reads\",\"signal_to_noise\",\"snr\",\"warning\",\"reactivity\",\"errors\",\"score_start_idx\",\"score_end_idx\"]\n",
    "        )\n",
    "\n",
    "        # Add the row to the whole dataframe\n",
    "        seqList.append(row)\n",
    "\n",
    "df = pd.concat(seqList, ignore_index=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "94537a77-c228-4989-9cac-dfc40b7982d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reset the index and sort by ID\n",
    "df.set_index('id', inplace=True)\n",
    "df.sort_values('id',ascending=False,inplace=True)\n",
    "\n",
    "# Persist the dataframe for loading in future steps\n",
    "# See the README for details about format choice\n",
    "\n",
    "# df.to_csv('../data/data_rdatOnly.csv')\n",
    "df.to_pickle('../data/data_rdatOnly.pkl')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "37b02c44-ec2e-42a2-867f-1cabd8380baa",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# OPTIONAL\n",
    "# Some sequences will have poor data quality, indicated with a badQuality warning\n",
    "# To save compute, you may choose to filter out the low quality sequences\n",
    "\n",
    "goodQuality = df.loc[df['warning'] != 'badQuality']\n",
    "\n",
    "# goodQuality.to_csv('../data/data_highQuality.csv')\n",
    "goodQuality.to_pickle('../data/data_highQuality.pkl')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "venv",
   "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.12.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}