Remove unused IlluminaFastqHeader and CheckAfterCompleteMixin

Author: arteymix

rnaseq_pipeline/utils.py

@@ -5,36 +5,6 @@
 
 logger = logging.getLogger(__name__)
 
-class IlluminaFastqHeader:
-    @classmethod
-    def parse(cls, s):
-        pieces = s.split(':')
-        if len(pieces) == 5:
-            device, flowcell_lane, tile, x, y = pieces
-            return cls(device, flowcell_lane=flowcell_lane, tile=tile, x=x, y=y)
-        elif len(pieces) == 7:
-            return cls(*pieces)
-        else:
-            raise TypeError('Unsupported Illumina FASTQ header format {}.'.format(s))
-
-    def __init__(self, device, run=None, flowcell=None, flowcell_lane=None, tile=None, x=None, y=None):
-        self.device = device
-        self.run = run
-        self.flowcell = flowcell
-        self.flowcell_lane = flowcell_lane
-        self.tile = tile
-        self.x = x
-        self.y = y
-
-    @property
-    def batch_factor(self):
-        if self.flowcell is None:
-            return self.device, self.flowcell_lane
-        return self.device, self.flowcell, self.flowcell_lane
-
-def parse_illumina_fastq_header(s):
-    return IlluminaFastqHeader(*s.split(':'))
-
 def max_retry(count):
     """
     Set the maximum number of time a task can be retried before being disabled
@@ -68,15 +38,6 @@ def run(self):
     def complete(self):
         return (not self.rerun or self._has_rerun) and super().complete()
 
-class CheckAfterCompleteMixin(luigi.Task):
-    """Ensures that a task is completed after a successful run()."""
-
-    def run(self):
-        ret = super().run()
-        if not self.complete():
-            raise RuntimeError('{} is not completed after successful run().'.format(repr(self)))
-        return ret
-
 def remove_task_output(task):
     logger.info('Cleaning up %s...', repr(task))
     for out in flatten_output(task):