2_ms2_annotation/sirius/0_ms2_into_groups.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import glob,os,re,sys,shutil\n",
    "from natsort import natsorted\n",
    "from itertools import zip_longest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1707 CHARGE=1-\n",
      " 155 CHARGE=2-\n",
      "   2 CHARGE=3-\n",
      "   1 CHARGE=4-\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "cat ../../1_preprocessing/ms2spectra_consensus.mgf | grep \"CHARGE\" | sort | uniq -c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "hideCode": false
   },
   "outputs": [],
   "source": [
    "###############################################\n",
    "# Functions:\n",
    "\n",
    "### get begin and end indices into tuples:\n",
    "def get_indices(file):\n",
    "    begin_line_indexes = []\n",
    "    end_line_indexes = []\n",
    "\n",
    "    with open(file,'r') as ms2_consensus:\n",
    "        for i, line in enumerate(ms2_consensus.readlines()):\n",
    "            if 'BEGIN' in line:\n",
    "            #   extract line index for lines that contain BEGIN\n",
    "                begin_line_indexes.append(i)\n",
    "\n",
    "            elif 'END' in line:\n",
    "                # return index of line after //\n",
    "                end_line_indexes.append(i+1)\n",
    "\n",
    "    begin_end_tuple = list(zip(begin_line_indexes, end_line_indexes))\n",
    "\n",
    "    return begin_end_tuple\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "#### grouped() groups n features into mgf files\n",
    "def grouper(iterable, n, fillvalue=None):\n",
    "    args = [iter(iterable)] * n\n",
    "    return zip_longest(*args, fillvalue=fillvalue)\n",
    "\n",
    "\n",
    "###############################################"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "file = \"../../1_preprocessing/ms2spectra_consensus.mgf\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "### 4 - create new dir to have separate mgf files\n",
    "dirname = 'groups'\n",
    "if os.path.exists(dirname):\n",
    "    shutil.rmtree(dirname)\n",
    "os.makedirs(dirname)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1865\n",
      "1563\n"
     ]
    }
   ],
   "source": [
    "### 5 - repeat steps 1,2\n",
    "begin_end_tuple = get_indices(file)\n",
    "\n",
    "print(len(begin_end_tuple))\n",
    "\n",
    "count = 0\n",
    "for b in begin_end_tuple:\n",
    "    if b[1]-b[0] < 9:\n",
    "        begin_end_tuple.pop(count)\n",
    "        count+=1\n",
    "    else:\n",
    "        count+=1\n",
    "\n",
    "print(len(begin_end_tuple))\n",
    "        \n",
    "\n",
    "lineList = list()\n",
    "with open(file, 'r') as f:\n",
    "    lineList = [line.rstrip() for line in f]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "### 6 - separate mgf consensus into n files with 5 ms2 spectra\n",
    "count = 1\n",
    "\n",
    "for group in list(grouper(begin_end_tuple , 5)):\n",
    "    with open(dirname+'/group_'+str(count)+'.mgf','w') as out:\n",
    "        if None in group:\n",
    "            none_index = group.index(None)\n",
    "            group = group[:none_index] # easy fix\n",
    "            for g in group:\n",
    "                ms2 = lineList[g[0]:g[1]]\n",
    "                for i in ms2:\n",
    "                    out.write(i+'\\n')\n",
    "                out.write('\\n')\n",
    "            count+=1\n",
    "        else:\n",
    "            for g in group:\n",
    "                ms2 = lineList[g[0]:g[1]]\n",
    "                for i in ms2:\n",
    "                    out.write(i+'\\n')\n",
    "                out.write('\\n')\n",
    "            count+=1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "celltoolbar": "Hide code",
  "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.9.0"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "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
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}