# Path: snakemakelib/bio/ngs/qc/qualimap.py
# class Qualimap(Results):
#     _keys = ['globals', 'coverage_per_contig']
# 
#     def __init__(self, *args, **kw):
#         super(Qualimap, self).__init__(*args, **kw)
# 
#     def _collect_globals(self, data, first, sample):
#         df_tmp = self.parse_data(data,
#                                  rs=("Globals", "Insert"),
#                                  skip=1, split=True,
#                                  columns=GLOBALS_COLUMNS,
#                                  dtype=float, sep=" = ")
#         df_tmp['value'] = [float(x.split(" ")[0].replace(",", ""))
#                            for x in df_tmp['value']]
#         df_tmp['Sample'] = sample
#         try:
#             if first:
#                 self['globals'] = df_tmp
#             else:
#                 self['globals'] = self['globals'].append(df_tmp, ignore_index=True)
#         except:
#             smllogger.warn("failed to append data to globals dataframe")
# 
#     def _collect_coverage_per_contig(self, data, first, sample):
#         df_tmp = self.parse_data(data,
#                                  rs=("Coverage per contig", None),
#                                  skip=2, split=True,
#                                  columns=COVERAGE_PER_CONTIG_COLUMNS,
#                                  dtype=float)
#         df_tmp["Sample"] = sample
#         try:
#             df_tmp['chrlen_percent'] = 100 * df_tmp['chrlen'] /\
#                 sum(df_tmp['chrlen'])
#             df_tmp['mapped_bases_percent'] = 100 * df_tmp['mapped_bases'] /\
#                 sum(df_tmp['mapped_bases'])
#         except:
#             smllogger.warn("coverage_per_contig: failed to normalize data")
#         try:
#             if first:
#                 self['coverage_per_contig'] = df_tmp
#             else:
#                 self['coverage_per_contig'] = self['coverage_per_contig'].append(
#                     df_tmp, ignore_index=True)
#         except:
#             smllogger.warn("failed to append data to coverage_per_contig dataframe")
# 
#     def _collect_results(self):
#         smllogger.info("Collecting results")
#         first = True
#         for (f, s) in zip(self._inputfiles, self._samples):
#             smllogger.debug("Reading input file {f} for sample {s}".format(f=f, s=s))
#             data = self.load_lines(f)
#             self._collect_globals(data, first, s)
#             self._collect_coverage_per_contig(data, first, s)
#             first = False
#         if self['globals'] is not None:
#             self['globals'] = self['globals'].pivot(
#                 index='Sample', columns='name', values='value')
#             self['globals']['number of unique reads'] = self['globals']['number of mapped reads']\
#                                                         - self['globals']['number of duplicated reads']


import unittest
import pandas as pd
from unittest.mock import patch
from snakemakelib.bio.ngs.qc.qualimap import Qualimap
        self.data = ['>>>>>>> Globals\n',
                     'number of windows = 10\n',
                     'number of reads = 10,000,000\n',
                     'number of mapped reads = 9,900,000 (99.00%)\n',
                     'number of duplicated reads = 400,000\n\n',
                     '>>>>>>> Insert size\n',
                     '>>>>>>> Coverage per contig\n',
                     '\n',
                     '\t'.join(['foo', '11', '12', '1.1', '1.2\n']),
                     '\t'.join(['bar', '21', '22', '2.1', '2.2\n'])]

    @patch('snakemakelib.results.Results.load_lines')
    def test_qualimap_coverage(self, mock_load_lines):
        mock_load_lines.return_value = self.data
        qm = Qualimap([('foo', 'bar')])
        self.assertListEqual([34.375, 65.625],
                             list(qm['coverage_per_contig']['chrlen_percent']))

    @patch('snakemakelib.results.Results.load_lines')
    def test_qualimap_globals(self, mock_load_lines):
        mock_load_lines.return_value = self.data
        qm = Qualimap([('foo', 'bar')])
        self.assertListEqual(sorted([10.0, 400000.0, 9500000.0, 10000000.0, 9900000.0]),
                             sorted(list(qm['globals'].loc['bar'])))

    @patch('pandas.DataFrame')
    @patch('snakemakelib.results.Results.load_lines')
    def test_collect_results(self, mock_load_lines, mock_df):
        mock_load_lines.return_value = self.data
        mock_df.return_value = pd.DataFrame()

# Path: snakemakelib/bio/ngs/qc/qualimap.py
# class Qualimap(Results):
#     _keys = ['globals', 'coverage_per_contig']
# 
#     def __init__(self, *args, **kw):
#         super(Qualimap, self).__init__(*args, **kw)
# 
#     def _collect_globals(self, data, first, sample):
#         df_tmp = self.parse_data(data,
#                                  rs=("Globals", "Insert"),
#                                  skip=1, split=True,
#                                  columns=GLOBALS_COLUMNS,
#                                  dtype=float, sep=" = ")
#         df_tmp['value'] = [float(x.split(" ")[0].replace(",", ""))
#                            for x in df_tmp['value']]
#         df_tmp['Sample'] = sample
#         try:
#             if first:
#                 self['globals'] = df_tmp
#             else:
#                 self['globals'] = self['globals'].append(df_tmp, ignore_index=True)
#         except:
#             smllogger.warn("failed to append data to globals dataframe")
# 
#     def _collect_coverage_per_contig(self, data, first, sample):
#         df_tmp = self.parse_data(data,
#                                  rs=("Coverage per contig", None),
#                                  skip=2, split=True,
#                                  columns=COVERAGE_PER_CONTIG_COLUMNS,
#                                  dtype=float)
#         df_tmp["Sample"] = sample
#         try:
#             df_tmp['chrlen_percent'] = 100 * df_tmp['chrlen'] /\
#                 sum(df_tmp['chrlen'])
#             df_tmp['mapped_bases_percent'] = 100 * df_tmp['mapped_bases'] /\
#                 sum(df_tmp['mapped_bases'])
#         except:
#             smllogger.warn("coverage_per_contig: failed to normalize data")
#         try:
#             if first:
#                 self['coverage_per_contig'] = df_tmp
#             else:
#                 self['coverage_per_contig'] = self['coverage_per_contig'].append(
#                     df_tmp, ignore_index=True)
#         except:
#             smllogger.warn("failed to append data to coverage_per_contig dataframe")
# 
#     def _collect_results(self):
#         smllogger.info("Collecting results")
#         first = True
#         for (f, s) in zip(self._inputfiles, self._samples):
#             smllogger.debug("Reading input file {f} for sample {s}".format(f=f, s=s))
#             data = self.load_lines(f)
#             self._collect_globals(data, first, s)
#             self._collect_coverage_per_contig(data, first, s)
#             first = False
#         if self['globals'] is not None:
#             self['globals'] = self['globals'].pivot(
#                 index='Sample', columns='name', values='value')
#             self['globals']['number of unique reads'] = self['globals']['number of mapped reads']\
#                                                         - self['globals']['number of duplicated reads']


# Path: snakemakelib/bio/ngs/tests/test_qualimap.py
import unittest
import pandas as pd
from unittest.mock import patch
from snakemakelib.bio.ngs.qc.qualimap import Qualimap

        self.data = ['>>>>>>> Globals\n',
                     'number of windows = 10\n',
                     'number of reads = 10,000,000\n',
                     'number of mapped reads = 9,900,000 (99.00%)\n',
                     'number of duplicated reads = 400,000\n\n',
                     '>>>>>>> Insert size\n',
                     '>>>>>>> Coverage per contig\n',
                     '\n',
                     '\t'.join(['foo', '11', '12', '1.1', '1.2\n']),
                     '\t'.join(['bar', '21', '22', '2.1', '2.2\n'])]

    @patch('snakemakelib.results.Results.load_lines')
    def test_qualimap_coverage(self, mock_load_lines):
        mock_load_lines.return_value = self.data
        qm = Qualimap([('foo', 'bar')])
        self.assertListEqual([34.375, 65.625],
                             list(qm['coverage_per_contig']['chrlen_percent']))

    @patch('snakemakelib.results.Results.load_lines')
    def test_qualimap_globals(self, mock_load_lines):
        mock_load_lines.return_value = self.data
        qm = Qualimap([('foo', 'bar')])
        self.assertListEqual(sorted([10.0, 400000.0, 9500000.0, 10000000.0, 9900000.0]),
                             sorted(list(qm['globals'].loc['bar'])))

    @patch('pandas.DataFrame')
    @patch('snakemakelib.results.Results.load_lines')
    def test_collect_results(self, mock_load_lines, mock_df):
        mock_load_lines.return_value = self.data
        mock_df.return_value = pd.DataFrame()

        Qualimap([('foo', 'bar')])