Source code for pyGenClean.RelatedSamples.merge_related_samples

#!/usr/bin/env python2.7

# This file is part of pyGenClean.
#
# pyGenClean is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# pyGenClean is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# pyGenClean.  If not, see <http://www.gnu.org/licenses/>.


import os
import sys
import gzip
import random
import logging
import argparse

from .. import __version__


logger = logging.getLogger("merge_related_samples")


[docs]def main(argString=None): """The main function of the module. :param argString: the options. :type argString: list """ # Getting and checking the options args = parseArgs(argString) checkArgs(args) merge_related_samples(args.ibs_related, args.out, args.no_status)
[docs]def checkArgs(args): """Checks the arguments and options. :param args: a an object containing the options of the program. :type args: argparse.Namespace :returns: ``True`` if everything was OK. If there is a problem with an option, an exception is raised using the :py:class:`ProgramError` class, a message is printed to the :class:`sys.stderr` and the program exists with code 1. """ if not os.path.isfile(args.ibs_related): msg = "{}: no such file".format(args.ibs_related) raise ProgramError(msg) return True
[docs]def parseArgs(argString=None): # pragma: no cover """Parses the command line options and arguments. :param argString: the options. :type argString: list :returns: A :py:class:`argparse.Namespace` object created by the :py:mod:`argparse` module. It contains the values of the different options. ================= ====== ================================================ Options Type Description ================= ====== ================================================ ``--ibs-related`` string The input file containing related individuals according to IBS value. ``--no-status`` bool The input file doesn't have a ``status`` column. ``--out`` string The prefix of the output files. ================= ====== ================================================ .. note:: No option check is done here (except for the one automatically done by argparse). Those need to be done elsewhere (see :py:func:`checkArgs`). """ args = None if argString is None: args = parser.parse_args() else: args = parser.parse_args(argString) return args
[docs]class ProgramError(Exception): """An :py:class:`Exception` raised in case of a problem. :param msg: the message to print to the user before exiting. :type msg: str """ def __init__(self, msg): """Construction of the :py:class:`ProgramError` class. :param msg: the message to print to the user :type msg: str """ self.message = str(msg) def __str__(self): return self.message
# The parser object desc = "Merges related samples according to IBS." long_desc = None parser = argparse.ArgumentParser(description=desc) parser.add_argument("-v", "--version", action="version", version="pyGenClean version {}".format(__version__)) # The INPUT files group = parser.add_argument_group("Input File") group.add_argument("--ibs-related", type=str, metavar="FILE", required=True, help=("The input file containing related individuals " "according to IBS value.")) # The options group = parser.add_argument_group("Options") group.add_argument("--no-status", action="store_true", help="The input file doesn't have a 'status' column.") # The OUTPUT files group = parser.add_argument_group("Output File") group.add_argument("--out", type=str, metavar="FILE", default="ibs_merged", help=("The prefix of the output files. [default: " "%(default)s]"))
[docs]def safe_main(): """A safe version of the main function (that catches ProgramError).""" try: main() except KeyboardInterrupt: logger.info("Cancelled by user") sys.exit(0) except ProgramError as e: logger.error(e.message) parser.error(e.message)
if __name__ == "__main__": safe_main()