feat (parseSTARFusion): added --skip-failed

Author: zhuchcn

CHANGELOG.md

@@ -12,7 +12,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [1.4.6-rc3] - 2025-03-10
 
-- Added --skip-failed flag to callVariant, parseArriba.
+- Added --skip-failed flag to callVariant, parseArriba, parserSTARFusion.
 
 ## [1.4.6-rc2] - 2025-03-03
 

moPepGen/cli/parse_star_fusion.py

@@ -3,12 +3,16 @@
 GVF file. The GVF file can be later used to call variant peptides using
 [callVariant](call-variant.md)."""
 from __future__ import annotations
+from typing import TYPE_CHECKING
 import argparse
 from typing import List
 from moPepGen import get_logger, seqvar, parser, err
 from moPepGen.cli import common
 
 
+if TYPE_CHECKING:
+    from logging import Logger
+
 INPUT_FILE_FORMATS = ['.tsv', '.txt']
 OUTPUT_FILE_FORMATS = ['.gvf']
 
@@ -37,16 +41,48 @@ def add_subparser_parse_star_fusion(subparsers:argparse._SubParsersAction):
         default=5.0,
         metavar='<number>'
     )
+    common.add_args_skip_failed(p)
     common.add_args_source(p)
     common.add_args_reference(p, proteome=False)
     common.add_args_debug_level(p)
     p.set_defaults(func=parse_star_fusion)
     common.print_help_if_missing_args(p)
     return p
 
+class TallyTable():
+    """ Tally table """
+    def __init__(self, logger:Logger):
+        """ Constructor """
+        self.total:int = 0
+        self.succeed:int = 0
+        self.skipped:TallyTableSkipped = TallyTableSkipped()
+        self.logger = logger
+
+    def log(self):
+        """ Show tally results """
+        self.logger.info("Summary:")
+        self.logger.info("Totally records read: %i", self.total)
+        self.logger.info("Records successfully processed: %i", self.succeed)
+        self.logger.info("Records skipped: %i", self.skipped.total)
+        if self.skipped.total > 0:
+            self.logger.info("Out of those skipped,")
+            self.logger.info("    Invalid gene ID: %i", self.skipped.invalid_gene_id)
+            self.logger.info("    Invalid position: %i", self.skipped.invalid_position)
+            self.logger.info("    Insufficient evidence: %i", self.skipped.insufficient_evidence)
+
+class TallyTableSkipped():
+    """ Tally table for failed ones """
+    def __init__(self):
+        """ constructor """
+        self.invalid_gene_id:int = 0
+        self.invalid_position:int = 0
+        self.insufficient_evidence:int = 0
+        self.total:int = 0
+
 def parse_star_fusion(args:argparse.Namespace) -> None:
     """ Parse the STAR-Fusion's output and save it in GVF format. """
     logger = get_logger()
+    tally = TallyTable(logger)
     # unpack args
     fusion = args.input_path
     common.validate_file_format(
@@ -64,13 +100,23 @@ def parse_star_fusion(args:argparse.Namespace) -> None:
     variants:List[seqvar.VariantRecord] = []
 
     for record in parser.STARFusionParser.parse(fusion):
+        tally.total += 1
         if record.est_j < args.min_est_j:
+            tally.skipped.insufficient_evidence += 1
             continue
         try:
             var_records = record.convert_to_variant_records(anno, genome)
+            variants.extend(var_records)
+            tally.succeed += 1
         except err.GeneNotFoundError:
+            tally.skipped.invalid_gene_id += 1
             continue
-        variants.extend(var_records)
+        except:
+            if args.skip_failed:
+                tally.skipped.invalid_position += 1
+                continue
+            else:
+                raise
 
     logger.info('STAR-Fusion output %s loaded.', fusion)
 
@@ -88,3 +134,5 @@ def parse_star_fusion(args:argparse.Namespace) -> None:
     seqvar.io.write(variants, output_path, metadata)
 
     logger.info('Variant info written to disk.')
+
+    tally.log()

test/integration/test_parse_star_fusion.py

@@ -2,6 +2,7 @@
 import argparse
 import subprocess as sp
 import sys
+from unittest.mock import Mock
 from test.integration import TestCaseIntegration
 from moPepGen import cli, seqvar
 from moPepGen.cli.common import load_references
@@ -27,19 +28,25 @@ def test_parse_star_fusion_cli(self):
             print(res.stderr.decode('utf-8'))
             raise
 
-    def test_star_fusion_record_case1(self):
-        """ Test parseSTARFusion """
+    def create_base_args(self) -> argparse.Namespace:
+        """ Create base args """
         args = argparse.Namespace()
         args.command = 'parseSTARFusion'
         args.source = 'Fusion'
-        args.input_path = self.data_dir/'fusion/star_fusion.txt'
         args.index_dir = None
         args.genome_fasta = self.data_dir/'genome.fasta'
         args.annotation_gtf = self.data_dir/'annotation.gtf'
         args.reference_source = None
         args.output_path = self.work_dir/'star_fusion.gvf'
         args.min_est_j = 3.0
         args.quiet = True
+        args.skip_failed =False
+        return args
+
+    def test_star_fusion_record_case1(self):
+        """ Test parseSTARFusion """
+        args = self.create_base_args()
+        args.input_path = self.data_dir/'fusion/star_fusion.txt'
         cli.parse_star_fusion(args)
         files = {str(file.name) for file in self.work_dir.glob('*')}
         expected = {'star_fusion.gvf'}
@@ -64,19 +71,23 @@ def test_star_fusion_record_case1(self):
 
     def test_parse_star_fusion_case1(self):
         """ test parseSTARFusion case1 """
-        args = argparse.Namespace()
-        args.command = 'parseSTARFusion'
+        args = self.create_base_args()
         args.input_path = self.data_dir/'fusion/star_fusion.txt'
-        args.source = 'Fusion'
-        args.index_dir = None
-        args.genome_fasta = self.data_dir/'genome.fasta'
-        args.annotation_gtf = self.data_dir/'annotation.gtf'
-        args.reference_source = None
-        args.output_path = self.work_dir/'star_fusion.gvf'
-        args.min_est_j = 3.0
-        args.quiet = True
         cli.parse_star_fusion(args)
         files = {str(file.name) for file in self.work_dir.glob('*')}
         expected = {'star_fusion.gvf'}
         self.assertEqual(files, expected)
         self.assert_gvf_order(args.output_path, args.annotation_gtf)
+
+    def test_parse_star_fusion_skip_failed(self):
+        """ test parseSTARFusion case1 """
+        from moPepGen import parser
+        parser.STARFusionParser.STARFusionRecord.convert_to_variant_records = Mock(
+            side_effect=ValueError()
+        )
+        args = self.create_base_args()
+        args.input_path = self.data_dir/'fusion/star_fusion.txt'
+        with self.assertRaises(ValueError):
+            cli.parse_star_fusion(args)
+        args.skip_failed = True
+        cli.parse_star_fusion(args)