hw18 add

bad_input.fastq

@@ -0,0 +1 @@
+This is a bad fastq input file

fasstq_filter.py

@@ -7,9 +7,39 @@ def GC(sequence):
     # Older versions have this:
     from Bio.SeqUtils import GC
 
+import argparse
+import logging
+
+logger = logging.getLogger('my_logger')
+logger.setLevel(logging.DEBUG)  # Capture all levels
+
+# Create formatters
+formatter = logging.Formatter('%(levelname)s - %(message)s')
+
+# Handler for DEBUG, INFO, and WARNING
+info_handler = logging.FileHandler('info.log')
+info_handler.setLevel(logging.DEBUG)
+info_handler.addFilter(lambda record: record.levelno <= logging.WARNING)
+info_handler.setFormatter(formatter)
+
+# Handler for ERROR and CRITICAL
+error_handler = logging.FileHandler('error.log')
+error_handler.setLevel(logging.ERROR)
+error_handler.setFormatter(formatter)
+
+# Add handlers to the logger
+logger.addHandler(info_handler)
+logger.addHandler(error_handler)
+
+
+
 def filter_fastq(input_fastq, output_fastq, gc_bounds=(0, 100), length_bounds=(0, 2 ** 32), quality_threshold=0):
+    if quality_threshold < 0:
+        logger.error(f"quality_threshold={quality_threshold}, can't be lower then 0, change to 0")
+        quality_threshold = 0
 
     if not isinstance(gc_bounds, tuple):
+        logger.info(f'You entered one number {gc_bounds}, it will be interpreted as the upper bound')
         gc_bounds = (0, gc_bounds)
 
     if not isinstance(length_bounds, tuple):
@@ -33,4 +63,41 @@ def filter_fastq(input_fastq, output_fastq, gc_bounds=(0, 100), length_bounds=(0
 
         filtered_records.append(record)
 
-    SeqIO.write(filtered_records, output_fastq, "fastq")
+    SeqIO.write(filtered_records, output_fastq, "fastq")
+
+
+def int_or_range(value):
+    try:
+        if ',' in value:
+            parts = value.split(',')
+            if len(parts) != 2:
+                raise ValueError
+            return tuple(int(x.strip()) for x in parts)
+        return int(value)
+    except Exception:
+        raise argparse.ArgumentTypeError(f"Expected int or int range (start,end), got '{value}'")
+
+
+def get_args():
+    parser = argparse.ArgumentParser(
+                        prog='My wonderful parser',
+                        description='This tool is needed to parse command line arguments',
+                        epilog='Text at the bottom of help')
+
+    parser.add_argument('-i', "--input", required=True, metavar='path', type=str, help='This is fastq format input file (takes value)')
+    parser.add_argument('-g', "--gc", type=int_or_range, default=(0, 100), help="Either a single integer (e.g., 5) or a range in the form start,end (e.g., 0,100)")
+    parser.add_argument('-l', "--lenb", type=int_or_range, default=(0,  2 ** 32), help="Either a single integer (e.g., 5) or a range in the form start,end (e.g., 0,  2 ** 32)")
+    parser.add_argument('-t', "--thresh", type=int, default=0, help='Threshold value of average read quality for filtering (default is 0, phred33 scale)')
+    parser.add_argument('-o', "--output", required=True, metavar='path', type=str, help='This is fastq format filtered output file')
+    return parser.parse_args()
+
+if __name__ == '__main__':
+    args = get_args()
+    # args.gggg
+    # print(args)
+    input_fastq=args.input
+    output_fastq=args.output
+    gc_bounds=args.gc
+    length_bounds=args.lenb
+    quality_threshold=args.thresh
+    filter_fastq(input_fastq, output_fastq, gc_bounds, length_bounds, quality_threshold)

fastq_test.py

@@ -0,0 +1,90 @@
+import os
+import pytest
+from fasstq_filter import filter_fastq
+
+def test_no_bound():
+    try:
+        filter_fastq("./input.fastq", "./no_bound_output.fastq")
+        with open('./input.fastq', 'r') as f1, open('./no_bound_output.fastq', 'r') as f2:
+            lines1 = f1.readlines()
+            lines2 = f2.readlines()
+        assert lines1 == lines2, "input file not same as output file"
+    finally:
+        if os.path.exists("./no_bound_output.fastq"):
+            os.remove("./no_bound_output.fastq")
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        ({"length_bounds": 1}),
+        ({"gc_bounds": 1}),
+        ({"quality_threshold": 101}),
+    ]
+)
+def test_bound_all(params):
+    try:
+        filter_fastq("./input.fastq", "./no_bound_all.fastq", **params)
+        with open('./no_bound_all.fastq', 'r') as f1:
+            lines1 = f1.readlines()
+        assert not lines1, "file not empty"
+    finally:
+        if os.path.exists("./no_bound_all.fastq"):
+            os.remove("./no_bound_all.fastq")
+
+
+@pytest.mark.parametrize(
+    "params_int, params_tuple",
+    [
+        ({"length_bounds": 50}, {"length_bounds": (0, 50)}),
+        ({"gc_bounds": 50}, {"gc_bounds": (0, 50)}),
+    ]
+)
+def test_int_tuple(params_int, params_tuple):
+    try:
+        filter_fastq("./input.fastq", "./test_int.fastq", **params_int)
+        filter_fastq("./input.fastq", "./test_tuple.fastq", **params_tuple)
+        with open("./test_int.fastq", 'r') as f1, open("./test_tuple.fastq", 'r') as f2:
+            lines1 = f1.readlines()
+            lines2 = f2.readlines()
+        assert lines1 == lines2, "int file not same as tuple file"
+    finally:
+        if os.path.exists("./test_int.fastq"):
+            os.remove("./test_int.fastq")
+        if os.path.exists("./test_tuple.fastq"):
+            os.remove("./test_tuple.fastq")
+
+
+def test_input_error():
+    try:
+        filter_fastq("./bad_input.fastq", "./bad_output_file.fastq")
+        assert "never come here"
+    except ValueError:
+        pass
+
+@pytest.mark.parametrize(
+    "length_bounds, gc_bounds, quality_threshold",
+    [
+        (0, 50, -11),
+    ]
+)
+def test_logger(length_bounds, gc_bounds, quality_threshold):
+    try:
+        filter_fastq("./input.fastq", "./outlog.fastq", length_bounds, gc_bounds, quality_threshold)
+        with open('./error.log', 'r') as f1:
+            lines1 = f1.readlines()
+        assert lines1[-1] == "ERROR - quality_threshold=-11, can't be lower then 0, change to 0\n", "wrong log"
+    finally:
+        if os.path.exists("./outlog.fastq"):
+            os.remove("./outlog.fastq")
+
+
+def test_logger():
+    try:
+        filter_fastq("./input.fastq", "./outlog.fastq", 0, 50, -11)
+        with open('./error.log', 'r') as f1:
+            lines1 = f1.readlines()
+        assert lines1[-1] == "ERROR - quality_threshold=-11, can't be lower then 0, change to 0\n", "wrong log"
+    finally:
+        if os.path.exists("./outlog.fastq"):
+            os.remove("./outlog.fastq")