234 lines
7.6 KiB
Python
234 lines
7.6 KiB
Python
"""Required sample sheet validators.
|
|
|
|
These should be applied to sample sheets from all workflows.
|
|
"""
|
|
|
|
from abc import ABC
|
|
import re
|
|
|
|
|
|
# Define the base class for sample sheet validators
|
|
class SampleSheetValidator(ABC):
|
|
"""Base class for sample sheet validators."""
|
|
|
|
def __init__(self, wf_params, options=None):
|
|
"""Initialize the validator with workflow parameters."""
|
|
self.wf_params = wf_params
|
|
self.options = options or {}
|
|
self.alias_field = "sample_name" if self.options.get("no_barcode") else "alias"
|
|
self.errors = []
|
|
|
|
def log_error(self, msg, column=None, lineno=None):
|
|
"""Log an error message with optional column and line context."""
|
|
context = []
|
|
if column is not None:
|
|
context.append(f"column: {column}")
|
|
if lineno is not None:
|
|
context.append(f"line: {lineno}")
|
|
if context:
|
|
msg = f"{msg} ({', '.join(context)})"
|
|
self.errors.append(msg)
|
|
|
|
def on_header(self, header):
|
|
"""Handle header line."""
|
|
pass
|
|
|
|
def add_sheet_row(self, row, lineno):
|
|
"""Handle a single row from the sample sheet."""
|
|
pass
|
|
|
|
@property
|
|
def is_valid(self):
|
|
"""Check if the sample sheet is valid according to this validator."""
|
|
if self.errors:
|
|
return False
|
|
return True
|
|
|
|
|
|
class ProhibitedColumns(SampleSheetValidator):
|
|
"""Check for prohibited columns."""
|
|
|
|
def __init__(self, wf_params, options=None):
|
|
"""Initialize with workflow parameters."""
|
|
super().__init__(wf_params, options)
|
|
|
|
def on_header(self, header):
|
|
"""Don't allow `barcode` and `alias` when `--no_barcode` set."""
|
|
if self.options.get("no_barcode"):
|
|
for field in ["alias", "barcode"]:
|
|
if field in header:
|
|
self.log_error(
|
|
"Column must not be present with --no_barcode",
|
|
column=field,
|
|
)
|
|
|
|
|
|
class RequiredColumns(SampleSheetValidator):
|
|
"""Check for required columns."""
|
|
|
|
def on_header(self, header):
|
|
"""Check for required columns."""
|
|
self.header = header
|
|
|
|
required = [self.alias_field]
|
|
if not self.options.get("no_barcode"):
|
|
# in barcode mode required are alias_field and barcode
|
|
required.insert(0, "barcode")
|
|
|
|
for req in required:
|
|
if req not in header:
|
|
self.log_error("Column missing", column=req)
|
|
|
|
def add_sheet_row(self, row, lineno):
|
|
"""Check for consistent number of columns."""
|
|
if len(row) != len(self.header):
|
|
self.log_error("Unexpected number of cells in row", lineno=lineno)
|
|
|
|
|
|
class SampleTypeRules(SampleSheetValidator):
|
|
"""Check sample types."""
|
|
|
|
ALLOWED = {
|
|
"test_sample", "positive_control", "negative_control", "no_template_control"
|
|
}
|
|
|
|
def __init__(self, wf_params, options=None):
|
|
"""Initialize with workflow parameters."""
|
|
super().__init__(wf_params, options)
|
|
self.types = []
|
|
self.unexpected_types = []
|
|
self.required = wf_params.get("required_sample_types", [])
|
|
|
|
def add_sheet_row(self, row, lineno):
|
|
"""Collect sample types."""
|
|
t = row.get("type")
|
|
if t:
|
|
self.types.append(t)
|
|
if t not in self.ALLOWED:
|
|
self.unexpected_types.append((t, lineno))
|
|
|
|
@property
|
|
def is_valid(self):
|
|
"""Check if the sample types are valid."""
|
|
valid = True
|
|
if self.unexpected_types:
|
|
for t, lineno in self.unexpected_types:
|
|
self.log_error(
|
|
f"Unexpected value: {t}", column="type", lineno=lineno)
|
|
valid = False
|
|
for required in self.required:
|
|
if required not in self.ALLOWED:
|
|
self.log_error(f"Not an allowed sample type: {required}")
|
|
valid = False
|
|
elif required not in self.types:
|
|
self.log_error(f"Sample sheet requires at least 1 of '{required}'")
|
|
valid = False
|
|
return valid
|
|
|
|
|
|
class BarcodeRules(SampleSheetValidator):
|
|
"""Check barcode rules."""
|
|
|
|
def __init__(self, wf_params, options=None):
|
|
"""Initialize with workflow parameters."""
|
|
super().__init__(wf_params, options)
|
|
self.first_len = None
|
|
self.barcodes = []
|
|
|
|
def add_sheet_row(self, row, lineno):
|
|
"""Check barcode rules."""
|
|
if self.options.get("no_barcode"):
|
|
return
|
|
|
|
bc = row.get("barcode")
|
|
if bc is None:
|
|
self.log_error("Column missing", column="barcode")
|
|
return
|
|
if bc in self.barcodes:
|
|
self.log_error(
|
|
f"Value not unique: {bc}", column="barcode", lineno=lineno
|
|
)
|
|
self.barcodes.append(bc)
|
|
if not re.match(r"^barcode\d\d+$", bc):
|
|
self.log_error(
|
|
f"Value has incorrect format: {bc}",
|
|
column="barcode",
|
|
lineno=lineno,
|
|
)
|
|
if self.first_len is None:
|
|
self.first_len = len(bc)
|
|
elif len(bc) != self.first_len:
|
|
self.log_error(
|
|
f"Values are different lengths: {bc}",
|
|
column="barcode",
|
|
lineno=lineno,
|
|
)
|
|
|
|
|
|
class AliasRules(SampleSheetValidator):
|
|
"""Alias rules."""
|
|
|
|
ALIAS_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
|
|
|
|
def __init__(self, wf_params, options=None):
|
|
"""Initialize with workflow parameters."""
|
|
super().__init__(wf_params, options)
|
|
self.aliases = set()
|
|
|
|
def add_sheet_row(self, row, lineno):
|
|
"""Check alias rules."""
|
|
alias = row.get(self.alias_field)
|
|
if alias is None:
|
|
self.log_error("Column missing", column=self.alias_field)
|
|
return
|
|
if alias in self.aliases:
|
|
self.log_error(
|
|
f"Value not unique: {alias}", column=self.alias_field, lineno=lineno,
|
|
)
|
|
self.aliases.add(alias)
|
|
# Specific error message if empty "" to improve user error message
|
|
if not alias:
|
|
self.log_error(
|
|
f"Empty {self.alias_field}. "
|
|
"Allowed values start with letters or numbers "
|
|
"and may contain only letters, numbers, '.', '_' or '-'",
|
|
column=self.alias_field,
|
|
lineno=lineno,
|
|
)
|
|
if not self.ALIAS_PATTERN.match(alias):
|
|
self.log_error(
|
|
f"Invalid value {alias}. Allowed values start with letters or numbers "
|
|
"and may contain only letters, numbers, '.', '_' or '-'",
|
|
column=self.alias_field,
|
|
lineno=lineno,
|
|
)
|
|
if alias.startswith("barcode"):
|
|
self.log_error(
|
|
f"Value must not begin with 'barcode': {alias}",
|
|
column=self.alias_field,
|
|
lineno=lineno,
|
|
)
|
|
|
|
|
|
class AnalysisGroupCompleteness(SampleSheetValidator):
|
|
"""Analysis groups."""
|
|
|
|
def __init__(self, wf_params, options=None):
|
|
"""Initialize with workflow parameters."""
|
|
super().__init__(wf_params, options)
|
|
self.has_analysis_groups = False
|
|
|
|
def on_header(self, header):
|
|
"""Check for required columns."""
|
|
if "analysis_group" in header:
|
|
self.has_analysis_groups = True
|
|
|
|
def add_sheet_row(self, row, lineno):
|
|
"""Check analysis group completeness."""
|
|
if self.has_analysis_groups and not row.get("analysis_group"):
|
|
self.log_error(
|
|
"Column exists but needs values in each row",
|
|
column="analysis_group",
|
|
lineno=lineno,
|
|
)
|