Merge pull request #19 from nkongenelly/DATAOPS-1178_use_checkqc_handler

Using checkqc illumina parser for bclconvert

Author: nkongenelly

projman_filler/app.py

@@ -11,11 +11,13 @@
 from projman_filler.models.db_models import FlowcellRunfolder
 from projman_filler.bcl2fastq_run_stats_parser import Bcl2fastqRunStatsParser
 from projman_filler.interop_run_stats_parser import InteropRunStatsParser
+from projman_filler.qc_data_parser import QCDataParser
 
 from projman_filler.repositories.sample_results_repo import SampleResultRepo
 from projman_filler.repositories.flowcell_lane_results_repo import FlowcellLaneResultsRepo
 from projman_filler.repositories.flowcell_runfolder_repo import FlowcellRunfolderRepo
-
+from checkQC.qc_data import QCData
+from checkQC.config import ConfigFactory
 
 class App(object):
 
@@ -43,8 +45,8 @@ def delete_existing_flowcell_from_db(self, flowcell_name, force):
             return
 
         if force:
-            print("Found the specified runfolder in the db, but got a force option, so will proceed to "
-                  "delete it and insert new values.")
+            print("Found the specified runfolder in the db, but got a force "
+                "option, so will proceed to delete it and insert new values.")
             self.flowcell_lane_results_repo.delete_by_flowcell_name(flowcell_name)
             self.flowcell_runfolder_repo.delete_by_flowcell_name(flowcell_name)
             self.sample_results_repo.delete_by_flowcell_name(flowcell_name)
@@ -59,50 +61,131 @@ def insert_flowcell_runfolder_into_db(self, runfolder, flowcell_name):
                                                run_date=runfolder_date)
         self.flowcell_runfolder_repo.add(flowcell_runfolder)
 
-    def insert_runfolder_into_db(self, runfolder, bcl2fastq_stats_dir, force=False, atac_seq_mode=False, olink_mode=False):
-        if olink_mode:
-            print("Olink mode activated. Will read lane-level statistics from InterOp files instead of bcl2fastq Stats.json.")
+    def insert_runfolder_into_db(self, runfolder, bcl2fastq_stats_dir,
+                                 demultiplexer, ch_config_path, force=False,
+                                 atac_seq_mode=False, olink_mode=False):
+        """
+        Inserts runfolder data into the specific database based on the specified 
+            demultiplexer and mode.
+
+        :param runfolder (str): Path to the runfolder directory.
+        :param bcl2fastq_stats_dir (str): Subdirectory containing bcl2fastq statistics.
+        :param demultiplexer (str): Demultiplexer used (e.g 'bcl2fastq' or 'bclconvert').
+        :param ch_config_path (Path): Path to the checkQC configuration file
+        :param force (bool, optional): If True, existing flowcell data will be overwritten.
+        :param atac_seq_mode (bool, optional): If True, enables ATAC-seq specific processing.
+        :param olink_mode (bool, optional): If True, enables Olink-specific processing.
+        """
+        flowcell_name = None
+
+        if demultiplexer == "bcl2fastq":
+            flowcell_name = self._handle_bcl2fastq(
+                runfolder, bcl2fastq_stats_dir, force, atac_seq_mode
+            )
+        elif olink_mode:
+            print("Olink mode activated. Will read lane-level statistics from "
+                  "InterOp files instead of bcl2fastq Stats.json.")
             return self.insert_olink_runfolder_into_db(runfolder, force)
+        else:
+            flowcell_name = self._handle_other_demultiplexer(
+               runfolder, demultiplexer, ch_config_path, force
+            )
+
+        self.insert_flowcell_runfolder_into_db(runfolder, flowcell_name)
+
+    def _handle_bcl2fastq(self, runfolder, stats_dir, force, atac_seq_mode):
+        """
+        Handles runfolder processing for the bcl2fastq demultiplexer.
+
+        :param runfolder (str): Path to the runfolder directory.
+        :param stats_dir (str): Subdirectory containing bcl2fastq statistics.
+        :param force (bool): If True, existing flowcell data will be overwritten.
+        :param atac_seq_mode (bool): If True, enables ATAC-seq specific processing.
+
+        :return flowcell_name (str): The flowcell name extracted from the bcl2fastq 
+            statistics (after saving flowcel_lane and sample data in DB).
+        """
+        stats_path = os.path.join(runfolder, stats_dir)
+        bcl2fastq_stats = Bcl2fastqRunStatsParser(stats_path)
 
-        bcl2fastq_stats = Bcl2fastqRunStatsParser(os.path.join(runfolder, bcl2fastq_stats_dir))
         flowcell_name = bcl2fastq_stats.get_flowcell_name()
         reads_and_cycles = bcl2fastq_stats.get_reads_and_cycles()
         conversion_results = bcl2fastq_stats.get_conversion_results()
 
-        # Check if flowcell exists and should be overriden
         self.delete_existing_flowcell_from_db(flowcell_name, force)
 
-        # For atac-seq we run bcl2fastq with special parameters declaring
-        # that the second index should be interpreted as a non-index read.
-        # So we allow overriding the Interop list of non-index-reads with
-        # a custom list obtained from bcl2fastq stats. /ML 2021-09
         non_index_reads = None
         if atac_seq_mode:
-            print("ATAC-seq mode activated. Will re-map read numbers according to settings used by bcl2fastq.")
+            print("ATAC-seq mode activated. Will re-map read numbers according "
+                  "to settings used by bcl2fastq.")
             non_index_reads = bcl2fastq_stats.get_non_index_reads()
-
+        
         interop = InteropRunStatsParser(runfolder, non_index_reads)
-        lane_stats = calculate_lane_statistics(interop, flowcell_name, conversion_results)
+        lane_stats = calculate_lane_statistics(
+            interop, flowcell_name, conversion_results
+        )
         self.flowcell_lane_results_repo.add(list(lane_stats))
 
-        samplesheet_file = os.path.join(runfolder, "SampleSheet.csv")
-        samplesheet = Samplesheet(samplesheet_file)
-
-        sample_stats = calculate_sample_statistics(flowcell_name, conversion_results, reads_and_cycles, samplesheet)
+        samplesheet = Samplesheet(os.path.join(runfolder, "SampleSheet.csv"))
+        sample_stats = calculate_sample_statistics(
+            flowcell_name, conversion_results, reads_and_cycles, samplesheet
+        )
         self.sample_results_repo.add(list(sample_stats))
 
-        self.insert_flowcell_runfolder_into_db(runfolder, flowcell_name)
-
+        return flowcell_name
 
     def insert_olink_runfolder_into_db(self, runfolder, force=False):
+        """
+        Inserts runfolder data into the database using Olink-specific processing.
+
+        :param runfolder (str): Path to the runfolder directory.
+        :param force (bool, optional): If True, existing flowcell data will be 
+            overwritten. Defaults to False.
+        """
         interop = InteropRunStatsParser(runfolder)
         flowcell_name = interop.get_flowcell_name()
 
         # Check if flowcell exists and should be overriden
         self.delete_existing_flowcell_from_db(flowcell_name, force)
 
         conversion_results = interop.get_conversion_results()
-        lane_stats = calculate_lane_statistics(interop, flowcell_name, conversion_results)
+        lane_stats = calculate_lane_statistics(
+            interop, flowcell_name, conversion_results
+        )
 
         self.flowcell_lane_results_repo.add(list(lane_stats))
         self.insert_flowcell_runfolder_into_db(runfolder, flowcell_name)
+
+    def _handle_other_demultiplexer(self, runfolder, demultiplexer, ch_config_path,
+                                     force):
+        """
+        Handles runfolder processing for demultiplexers other than bcl2fastq.
+
+        :param runfolder (str): Path to the runfolder directory.
+        :param demultiplexer (str): Demultiplexer used (e.g., 'bclconvert').
+        :param ch_config_path (Path): Path to the checkQC configuration file
+        :param force (bool): If True, existing flowcell data will be overwritten.
+
+        :return flowcell_name (str): The flowcell name extracted from the runfolder 
+            (after saving flowcel_lane and sample data in DB).
+        """
+
+        checkqc_conf = ConfigFactory.from_config_path(ch_config_path)._config
+        qc_data = getattr(QCData, f"from_{demultiplexer}")(
+            runfolder_path=runfolder,
+            parser_config=(
+                checkqc_conf
+                .get("parser_configurations", {})
+                .get(f"from_{demultiplexer}", {})
+            )
+        )
+        qc_data_parser = QCDataParser(qc_data, runfolder)
+        flowcell_name = qc_data_parser.flowcell_id
+        self.delete_existing_flowcell_from_db(flowcell_name, force)
+
+        lane_results = qc_data_parser._build_lane_results()
+        sample_results = qc_data_parser._build_sample_results()
+        self.flowcell_lane_results_repo.add(lane_results)
+        self.sample_results_repo.add(sample_results)
+
+        return flowcell_name

projman_filler/cli.py

@@ -18,15 +18,30 @@
 @click.option('--debug', is_flag=True)
 @click.option('--atac-seq-mode', is_flag=True)
 @click.option('--olink-mode', is_flag=True)
-@click.option('-b', '--bcl2fastq-stats', default="Unaligned/Stats", type=click.Path())
+@click.option('-b', '--bcl2fastq-stats', default="Unaligned/Stats", 
+    type=click.Path()
+)
+@click.option('-d', '--demultiplexer', default ="bcl2fastq", 
+    type=click.Choice(['bcl2fastq', 'bclconvert'])
+)
+@click.option(
+    "--config",
+    default="/etc/arteria/checkqc_config/checkqc.config",
+    type=click.Path(exists=True, dir_okay=False),
+    help="Path to the checkQC configuration file",
+)
 @click.argument('runfolder', type=click.Path())
-def main(runfolder, force, atac_seq_mode, olink_mode, bcl2fastq_stats, debug):
+def main(runfolder, force, atac_seq_mode, olink_mode, bcl2fastq_stats, 
+         demultiplexer, config, debug):
     """Console script for projman_filler."""
     print("projman_filler v{}".format(projman_filler_version))
     try:
         db_connection_string = os.environ["PROJMAN_DB"]
         app = App(db_connection_string, debug)
-        app.insert_runfolder_into_db(runfolder, bcl2fastq_stats, force=force, atac_seq_mode=atac_seq_mode, olink_mode=olink_mode)
+        app.insert_runfolder_into_db(
+            runfolder, bcl2fastq_stats, demultiplexer, config, force=force, 
+            atac_seq_mode=atac_seq_mode, olink_mode=olink_mode
+        )
     except FlowcellAlreadyInDb:
         print("Flowcell was already present in db.")
         sys.exit(1)

projman_filler/interop_run_stats_parser.py

@@ -128,5 +128,4 @@ def _get_conversion_results(self) -> list:
 
             lanes.append(Lane(l, total_clusters_raw, total_clusters_pf))
         return lanes
-
-
+    

projman_filler/qc_data_parser.py

@@ -0,0 +1,108 @@
+import pandas as pd
+from math import isnan
+
+from checkQC.parsers.illumina import _read_interop_summary
+from projman_filler.models.db_models import FlowcellLaneResult, SampleResult
+
+
+class QCDataParser:
+    def __init__(self, qc_data, runfolder):
+        """
+        Initializes the QCDataParser with QC data.
+
+        :param qc_data: QC data object from checkQC parser.
+        :param runfolder: Path to the sequencing run folder.
+        """
+        self.run_summary, index_summary, run_info = _read_interop_summary(runfolder)
+        self.flowcell_id = run_info.flowcell_id()
+        self.samplesheet_df = pd.DataFrame(qc_data.samplesheet)
+        self.qc_data = qc_data
+
+
+    def _build_lane_results(self):
+        """
+        Constructs lane-level statistics from checkQC sequencing metrics results 
+            from illumina parser.
+
+        :return results (List[FlowcellLaneResult]): Lane-level statistics for non-index reads.
+        """
+        results = []
+        for lane_no, lane_data in self.qc_data.sequencing_metrics.items():
+            read_no = 0
+            for number, read_data in lane_data["reads"].items():
+                if read_data["is_index"]:
+                    continue
+                read_no += 1 # counts only non-index reads
+                cycles = self.run_summary.at(number - 1).read().total_cycles()
+                error_rate = None if read_data["mean_error_rate"] and \
+                        isnan(read_data["mean_error_rate"]) else \
+                        read_data["mean_error_rate"]
+
+                results.append(FlowcellLaneResult(
+                    flowcell_id=self.flowcell_id,
+                    lane_num=lane_no,
+                    read_num=read_no,
+                    raw_density=lane_data["raw_density"],
+                    pf_density=lane_data["pf_density"],
+                    error_rate=error_rate,
+                    pf_clusters=lane_data["total_reads_pf"],
+                    raw_clusters=lane_data["total_reads"],
+                    cycles=cycles,
+                    pct_q30=read_data["percent_q30"] / 100,
+                ))
+        return results
+    
+    def _build_sample_results(self):
+        """
+        Constructs sample-level statistics by combining sequencing metrics, 
+            sample sheet data results from the checkQC interop parser
+
+        :return results (List[SampleResult]): Sample-level statistics including 
+            quality scores, indexing accuracy, and library metadata.
+        """
+        results = []
+        for lane_no, lane_data in self.qc_data.sequencing_metrics.items():
+            for sample_data in lane_data["reads_per_sample"]:
+                sample_id = sample_data["sample_id"]
+                sample_row = self.samplesheet_df[
+                    (self.samplesheet_df['lane'] == lane_no) &
+                    (self.samplesheet_df['sample_id'] == sample_id)
+                ]
+
+                library_name = \
+                    sample_row['custom_description'].to_string(index=False).split(
+                        "LIBRARY_NAME:"
+                    )[-1].strip()
+                index1 = sample_row['index'].to_string(index=False)
+                index2 = sample_row['index2'].to_string(index=False)
+                sample_project = sample_row['sample_project'].to_string(index=False)
+
+                read_no = 0
+                for number, read_data in lane_data["reads"].items():
+                    if read_data["is_index"]:
+                        continue
+                    read_no += 1 # counts only non-index reads
+                    cycles = self.run_summary.at(number - 1).read().total_cycles()
+                    no_of_reads = len(
+                        [
+                            no 
+                            for no, read_data in lane_data["reads"].items()
+                            if not read_data["is_index"]
+                        ]
+                    )
+                    results.append(SampleResult(
+                        flowcell_id=self.flowcell_id,
+                        project_id=sample_project,
+                        sample_name="_".join(sample_id.split("_")[1:]),
+                        tag_seq=f"{index1}-{index2}" if index2 else index1,
+                        lane_num=lane_no,
+                        read_num=read_no,
+                        cycles=cycles,
+                        pct_lane=sample_data["percent_of_lane"],
+                        pf_clusters=float(sample_data["cluster_count"]/no_of_reads),
+                        pct_q30=sample_data["percent_q30"],
+                        pct_tag_err=100 - sample_data["percent_perfect_index_reads"],
+                        library_name=library_name,
+                        mean_q=sample_data["mean_q30"],
+                    ))
+        return results

setup.py

@@ -16,7 +16,10 @@
     'SQLAlchemy~=2.0.40',
     'pymssql~=2.3.4',
     'pandas~=2.2.3',
-    'numpy~=2.2.4'
+    'numpy~=2.2.4',
+    # CheckQC pinned installation be changed once the correct checkQV version is published to pypi
+    'checkQC@git+https://github.com/Molmed/checkQC.git@master#egg=checkQC',
+
 ]
 
 setup_requirements = [

tests/resources/bcl2fastq/200624_A00834_0183_BHMTFYTINY/200624_A00834_0183_BHMTFYDRXX_no_adapters.csv

@@ -0,0 +1,14 @@
+[Header],,,,,
+Date,6/24/2020,,,,
+Application,Illumina DRAGEN COVIDSeq Test Pipeline,,,,
+Instrument Type,NovaSeq6000,,,,
+Assay,Illumina COVIDSeq Test,,,,
+Index Adapters,IDT-ILMN DNA-RNA UDP Indexes ,,,,
+Chemistry,Amplicon,,,,
+,,,,,
+[Settings],,,,,,,
+,,,,,
+[Data],,,,,
+Lane,Sample_ID,Sample_Name,Sample_Project,Index_ID,index,index2,Description
+1,Sample_AB-1234-14574-Qiagen-IndexSet1-SP-Lane1,AB-1234-14574-Qiagen-IndexSet1-SP-Lane1,AB-1234,UDP0001,GAACTGAGCG,TCGTGGAGCG,LIBRARY_NAME:test
+2,Sample_AB-1234-14574-Qiagen-IndexSet1-SP-Lane2,AB-1234-14574-Qiagen-IndexSet1-SP-Lane2,AB-1234,UDP0001,GAACTGAGCG,TCGTGGAGCG,LIBRARY_NAME:test