diff options
Diffstat (limited to 'quality_control')
| -rw-r--r-- | quality_control/average.py | 12 | ||||
| -rw-r--r-- | quality_control/errors.py | 8 | ||||
| -rw-r--r-- | quality_control/headers.py | 39 | ||||
| -rw-r--r-- | quality_control/standard_error.py | 17 |
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.")) |