Common parse_log() method made default

sequence_processing_pipeline/ConvertJob.py

@@ -175,6 +175,7 @@ def run(self, callback=None):
         logging.info(f'Successful job: {job_info}')
 
     def parse_logs(self):
+        # overrides Job.parse_logs() w/tailored parse for specific logs.
         log_path = join(self.output_path, 'Logs')
         errors = join(log_path, 'Errors.log')
 

sequence_processing_pipeline/FastQCJob.py

@@ -6,7 +6,6 @@
 from functools import partial
 from json import dumps
 import logging
-import glob
 
 
 class FastQCJob(Job):
@@ -305,19 +304,3 @@ def _generate_job_script(self):
 
         with open(sh_details_fp, 'w') as f:
             f.write('\n'.join(self.commands))
-
-    def parse_logs(self):
-        log_path = join(self.output_path, 'logs')
-        files = sorted(glob.glob(join(log_path, '*.out')))
-        msgs = []
-
-        for some_file in files:
-            with open(some_file, 'r') as f:
-                msgs += [line for line in f.readlines()
-                         # note 'error' is not same
-                         # requirement as found in QCJob.
-                         # ('error:'). This is a very
-                         # generalized filter.
-                         if 'error' in line.lower()]
-
-        return [msg.strip() for msg in msgs]

sequence_processing_pipeline/Job.py

@@ -13,6 +13,7 @@
 from inspect import stack
 import re
 from collections import Counter
+from glob import glob
 
 
 # taken from https://jinja.palletsprojects.com/en/3.0.x/api/#jinja2.BaseLoader
@@ -126,7 +127,17 @@ def run(self):
         raise PipelineError("Base class run() method not implemented.")
 
     def parse_logs(self):
-        raise PipelineError("Base class parse_logs() method not implemented.")
+        # by default, look for anything to parse in the logs directory.
+        log_path = join(self.output_path, 'logs')
+        files = sorted(glob(join(log_path, '*')))
+        msgs = []
+
+        for some_file in files:
+            with open(some_file, 'r') as f:
+                msgs += [line for line in f.readlines()
+                         if 'error:' in line.lower()]
+
+        return [msg.strip() for msg in msgs]
 
     def _which(self, file_path, modules_to_load=None):
         """

sequence_processing_pipeline/NuQCJob.py

@@ -499,16 +499,3 @@ def _generate_job_script(self, max_bucket_size):
                                     pmls_path=self.pmls_path))
 
         return job_script_path
-
-    def parse_logs(self):
-        log_path = join(self.output_path, 'logs')
-        # sorted lists give predictable results
-        files = sorted(glob(join(log_path, '*.out')))
-        msgs = []
-
-        for some_file in files:
-            with open(some_file, 'r') as f:
-                msgs += [line for line in f.readlines()
-                         if 'error:' in line.lower()]
-
-        return [msg.strip() for msg in msgs]

sequence_processing_pipeline/SeqCountsJob.py

@@ -5,6 +5,7 @@
 from jinja2 import Environment
 from os import walk
 from json import dumps
+from glob import glob
 
 
 logging.basicConfig(level=logging.DEBUG)
@@ -101,8 +102,16 @@ def _generate_job_script(self):
         return job_script_path
 
     def parse_logs(self):
-        # TODO
-        pass
+        # overrides Job.parse_logs() w/tailored parse for specific logs.
+        files = sorted(glob(join(self.log_path, '*.err')))
+        msgs = []
+
+        for some_file in files:
+            with open(some_file, 'r') as f:
+                msgs += [line for line in f.readlines()
+                         if line.startswith("[E::stk_size]")]
+
+        return [msg.strip() for msg in msgs]
 
     def _aggregate_counts(self):
         def extract_metadata(fp):

sequence_processing_pipeline/TRIntegrateJob.py

@@ -8,7 +8,6 @@
 from metapool import load_sample_sheet
 from os import makedirs
 from shutil import copyfile
-from glob import glob
 
 
 logging.basicConfig(level=logging.DEBUG)
@@ -162,16 +161,3 @@ def _generate_job_script(self):
                 "output_dir": self.output_path}))
 
         return job_script_path
-
-    def parse_logs(self):
-        log_path = join(self.output_path, 'logs')
-        # sorted lists give predictable results
-        files = sorted(glob(join(log_path, '*.out')))
-        msgs = []
-
-        for some_file in files:
-            with open(some_file, 'r') as f:
-                msgs += [line for line in f.readlines()
-                         if 'error:' in line.lower()]
-
-        return [msg.strip() for msg in msgs]

sequence_processing_pipeline/TellReadJob.py

@@ -6,7 +6,6 @@
 from .Pipeline import Pipeline
 from .PipelineError import PipelineError
 from metapool import load_sample_sheet
-from glob import glob
 
 
 logging.basicConfig(level=logging.DEBUG)
@@ -173,16 +172,3 @@ def _generate_job_script(self):
             }))
 
         return job_script_path
-
-    def parse_logs(self):
-        log_path = join(self.output_path, 'logs')
-        # sorted lists give predictable results
-        files = sorted(glob(join(log_path, '*.out')))
-        msgs = []
-
-        for some_file in files:
-            with open(some_file, 'r') as f:
-                msgs += [line for line in f.readlines()
-                         if 'error:' in line.lower()]
-
-        return [msg.strip() for msg in msgs]