<commit_before>#pylint: disable=invalid-name, too-few-public-methods
from __future__ import print_function, absolute_import, division
from argparse import Namespace
from collections import OrderedDict
import datetime
import os
import pysam
import sys
from testfixtures.tempdirectory import TempDirectory
import time
import unittest
import zither.zither as zither

try:
    from StringIO import StringIO
except ImportError:
    from io import StringIO

class _BamFlag(object):
    #the read is paired in sequencing, no matter whether it is mapped in a pair
    PAIRED = 1
    # the read is mapped in a proper pair
    PROPER_PAIR = 2
    #the read itself is unmapped; conflictive with PROPER_PAIR
    UNMAP = 4
    #the mate is unmapped
    MUNMAP = 8
    #the read is mapped to the reverse strand
    REVERSE = 16
    #the mate is mapped to the reverse strand
    MREVERSE = 32
    #this is read1
    READ1 = 64
    #this is read2
    READ2 = 128
    #not primary alignment
    SECONDARY = 256
    #QC failure
    QCFAIL = 512
    #optical or PCR duplicate
    DUP = 1024
    #supplementary alignment
    SUPPLEMENTARY = 2048

def _absolute_base_dir(*path_components):
    base_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
    return os.path.join(base_path, *path_components)

def _create_file(path, filename, contents):
    filename = os.path.join(path, filename)
    with open(filename, 'w') as new_file:
        new_file.write(contents)
    return filename

def _create_bam(path, filename, sam_contents):
    sam_filename = _create_file(path, filename, sam_contents)
    bam_filename = sam_filename.replace(".sam", ".bam")
    _pysam_bam_from_sam(sam_filename, bam_filename)
    return bam_filename

def _pysam_bam_from_sam(sam_filename, bam_filename):
    # I would rather just say pysam.view(...), but since that global is
    # added dynamically, Eclipse flags this as a compilation problem. So
    # instead we connect directly to the pysam.SamtoolsDispatcher.
    pysam_view = pysam.SamtoolsDispatcher("view", None).__call__
    pysam_index = pysam.SamtoolsDispatcher("index", None).__call__
    temp_stdout = sys.stdout
    try:
        sys.stdout = sys.__stdout__
        bam = pysam_view("-S", "-b", sam_filename)
        bam_file = open(bam_filename, "wb")
        for line in bam:
            bam_file.write(line)
        bam_file.close()
        pysam_index(bam_filename)

    finally:
        sys.stdout = temp_stdout

def _get_zither_metaheader(lines):
    return _get_line_starts_with(lines, "##zither=<")

def _get_column_header(lines):
    return _get_line_starts_with(lines, "#CHROM")

def _get_line_starts_with(lines, starts_with):
    for line in lines:
        if line.startswith(starts_with):
            return line
    return None


class ZitherBaseTestCase(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        self.simple_forward_read = (_BamFlag.PAIRED
                                    + _BamFlag.PROPER_PAIR
                                    + _BamFlag.MREVERSE
                                    + _BamFlag.READ1)
        self.stdout = StringIO()
        self.saved_stdout = sys.stdout
        sys.stdout = self.stdout<commit_msg>zith-12 (cgates): bughunt; adjust to pass in travis<commit_after>#pylint: disable=invalid-name, too-few-public-methods
from __future__ import print_function, absolute_import, division
from argparse import Namespace
from collections import OrderedDict
import datetime
import os
import pysam
import sys
from testfixtures.tempdirectory import TempDirectory
import time
import unittest
import zither.zither as zither

try:
    from StringIO import StringIO
except ImportError:
    from io import StringIO

# I would rather just say pysam.view(...), but since that global is
# added dynamically, Eclipse flags this as a compilation problem. So
# instead we connect directly to the pysam.SamtoolsDispatcher.
PYSAM_VIEW = pysam.SamtoolsDispatcher("view", None).__call__
PYSAM_INDEX = pysam.SamtoolsDispatcher("index", None).__call__

def _absolute_base_dir(*path_components):
    base_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
    return os.path.join(base_path, *path_components)

def _create_file(path, filename, contents):
    filename = os.path.join(path, filename)
    with open(filename, 'w') as new_file:
        new_file.write(contents)
    return filename

def _create_bam(path, filename, sam_contents):
    sam_filename = _create_file(path, filename, sam_contents)
    bam_filename = sam_filename.replace(".sam", ".bam")
    _pysam_bam_from_sam(sam_filename, bam_filename)
    return bam_filename

def _pysam_bam_from_sam(sam_filename, bam_filename):
    temp_stdout = sys.stdout
    try:
        sys.stdout = sys.__stdout__
        bam = PYSAM_VIEW("-S", "-b", sam_filename)
        bam_file = open(bam_filename, "wb")
        for line in bam:
            bam_file.write(line)
        bam_file.close()
        PYSAM_INDEX(bam_filename)

    finally:
        sys.stdout = temp_stdout

def _get_zither_metaheader(lines):
    return _get_line_starts_with(lines, "##zither=<")

def _get_column_header(lines):
    return _get_line_starts_with(lines, "#CHROM")

def _get_line_starts_with(lines, starts_with):
    for line in lines:
        if line.startswith(starts_with):
            return line
    return None


class _BamFlag(object):
    #the read is paired in sequencing, no matter whether it is mapped in a pair
    PAIRED = 1
    # the read is mapped in a proper pair
    PROPER_PAIR = 2
    #the read itself is unmapped; conflictive with PROPER_PAIR
    UNMAP = 4
    #the mate is unmapped
    MUNMAP = 8
    #the read is mapped to the reverse strand
    REVERSE = 16
    #the mate is mapped to the reverse strand
    MREVERSE = 32
    #this is read1
    READ1 = 64
    #this is read2
    READ2 = 128
    #not primary alignment
    SECONDARY = 256
    #QC failure
    QCFAIL = 512
    #optical or PCR duplicate
    DUP = 1024
    #supplementary alignment
    SUPPLEMENTARY = 2048


class ZitherBaseTestCase(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        self.simple_forward_read = (_BamFlag.PAIRED
                                    + _BamFlag.PROPER_PAIR
                                    + _BamFlag.MREVERSE
                                    + _BamFlag.READ1)
        self.stdout = StringIO()
        self.saved_stdout = sys.stdout
        sys.stdout = self.stdout
{'fraction_non_alphanumeric': 0.07624497370862976, 'fraction_numerical': 0.010207237859573151, 'mean_word_length': 3.009299442033478, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5069137', 'n_tokens_mistral': 2118, 'n_tokens_neox': 2027, 'n_words': 560}