aboutsummaryrefslogtreecommitdiff
path: root/quality_control
diff options
context:
space:
mode:
Diffstat (limited to 'quality_control')
-rw-r--r--quality_control/average.py12
-rw-r--r--quality_control/errors.py8
-rw-r--r--quality_control/headers.py39
-rw-r--r--quality_control/standard_error.py17
4 files changed, 76 insertions, 0 deletions
diff --git a/quality_control/average.py b/quality_control/average.py
index 2907b9c..9ca16a9 100644
--- a/quality_control/average.py
+++ b/quality_control/average.py
@@ -1,6 +1,8 @@
"""Contain logic for checking average files"""
import re
+from typing import Union
+from .errors import InvalidValue
from .errors import InvalidCellValue
def valid_value(val):
@@ -11,3 +13,13 @@ def valid_value(val):
f"Invalid value '{val}'. "
"Expected string representing a number with exactly three decimal "
"places.")
+
+def invalid_value(line_number: int, column_number: int, val: str) -> Union[
+ InvalidValue, None]:
+ if re.search(r"^[0-9]+\.[0-9]{3}$", val):
+ return None
+ return InvalidValue(
+ line_number, column_number, val, (
+ f"Invalid value '{val}'. "
+ "Expected string representing a number with exactly three decimal "
+ "places."))
diff --git a/quality_control/errors.py b/quality_control/errors.py
index 29a38f9..1eda646 100644
--- a/quality_control/errors.py
+++ b/quality_control/errors.py
@@ -1,5 +1,7 @@
"""Hold exceptions for QC package"""
+from collections import namedtuple
+
class InvalidCellValue(Exception):
"""Raised when a function encounters an invalid value"""
@@ -22,3 +24,9 @@ class ParseError(Exception):
"""Raised if any of the above exceptions are raised"""
def __init__(self, *args):
super().__init__(*args)
+
+InvalidValue = namedtuple(
+ "InvalidValue", ("line_number", "column_number", "value", "message"))
+
+DuplicateHeading = namedtuple(
+ "InvalidValue", ("line_number", "heading", "columns","message"))
diff --git a/quality_control/headers.py b/quality_control/headers.py
index b7bc01e..a5a5065 100644
--- a/quality_control/headers.py
+++ b/quality_control/headers.py
@@ -1,5 +1,9 @@
"""Validate the headers"""
+from functools import reduce
+from typing import Union, Tuple, Sequence
+
+from quality_control.errors import InvalidValue, DuplicateHeading
from quality_control.errors import DuplicateHeader, InvalidHeaderValue
def valid_header(strains, headers):
@@ -23,3 +27,38 @@ def valid_header(strains, headers):
for header, times in repeated))
return headers
+
+
+def invalid_header(
+ line_number: int, headers: Sequence[str]) -> Union[InvalidValue, None]:
+ if len(headers) < 2:
+ return InvalidValue(
+ line_number, 0, "<TAB>".join(headers),
+ "The header MUST contain at least 2 columns")
+
+def invalid_headings(
+ line_number: int, strains: Sequence[str],
+ headings: Sequence[str]) -> Union[Tuple[InvalidValue, ...], None]:
+ return tuple(
+ InvalidValue(
+ line_number, col, header, f"'{header}' not a valid strain.")
+ for col, header in
+ enumerate(headings, start=2) if header not in strains)
+
+def duplicate_headings(
+ line_number: int, headers: Sequence[str]) -> Union[InvalidValue, None]:
+ def __update_columns__(acc, item):
+ if item[1] in acc.keys():
+ return {**acc, item[1]: acc[item[1]] + (item[0],)}
+ return {**acc, item[1]: (item[0],)}
+ repeated = {
+ heading: columns for heading, columns in
+ reduce(__update_columns__, enumerate(headers, start=1), dict()).items()
+ if len(columns) > 1
+ }
+ return tuple(
+ DuplicateHeading(
+ line_number, heading, columns, (
+ f"Heading '{heading}', is repeated in columns "
+ f"{','.join(str(i) for i in columns)}"))
+ for heading, columns in repeated.items())
diff --git a/quality_control/standard_error.py b/quality_control/standard_error.py
index f1e33c4..022cc9b 100644
--- a/quality_control/standard_error.py
+++ b/quality_control/standard_error.py
@@ -1,6 +1,8 @@
"""Contain logic for checking standard error files"""
import re
+from typing import Union
+from .errors import InvalidValue
from .errors import InvalidCellValue
def valid_value(val):
@@ -11,3 +13,18 @@ def valid_value(val):
f"Invalid value '{val}'. "
"Expected string representing a number with at least six decimal "
"places.")
+
+def invalid_value(line_number: int, column_number: int, val: str) -> Union[
+ InvalidValue, None]:
+ """
+ Returns a `quality_control.errors.InvalidValue` object in the case where
+ `val` is not a valid input for standard error files, otherwise, it returns
+ `None`.
+ """
+ if re.search(r"^[0-9]+\.[0-9]{6,}$", val):
+ return None
+ return InvalidValue(
+ line_number, column_number, val, (
+ f"Invalid value '{val}'. "
+ "Expected string representing a number with at least six decimal "
+ "places."))