Use new sample sheet validation

This commit is contained in:
Natalia Garcia 2026-05-22 12:58:17 +00:00
parent 1b3155d591
commit 2cfc7b2549
2 changed files with 36 additions and 54 deletions

View File

@ -39,12 +39,12 @@ def determine_codec(f):
return None # will cause file to be opened with default encoding return None # will cause file to be opened with default encoding
def load_validators(modules, wf_params): def load_validators(modules, wf_params, options=None):
"""Load validator classes.""" """Load validator classes."""
validator_classes = [] validator_classes = []
for mod in modules: for mod in modules:
for _, validator_class in getmembers(mod, _is_sample_sheet_validator): for _, validator_class in getmembers(mod, _is_sample_sheet_validator):
validator_classes.append(validator_class(wf_params)) validator_classes.append(validator_class(wf_params, options))
return validator_classes return validator_classes
@ -76,11 +76,11 @@ def main(args):
if args.required_sample_types: if args.required_sample_types:
wf_params['required_sample_types'] = args.required_sample_types wf_params['required_sample_types'] = args.required_sample_types
# `no_barcode` is currently passed as a CLI option, validator_classes = load_validators(
# not through the workflow params JSON. load_validator_modules(), wf_params, options={
wf_params['no_barcode'] = args.no_barcode "no_barcode": args.no_barcode,
}
validator_classes = load_validators(load_validator_modules(), wf_params) )
rows = [] rows = []

View File

@ -11,9 +11,11 @@ import re
class SampleSheetValidator(ABC): class SampleSheetValidator(ABC):
"""Base class for sample sheet validators.""" """Base class for sample sheet validators."""
def __init__(self, wf_params): def __init__(self, wf_params, options=None):
"""Initialize the validator with workflow parameters.""" """Initialize the validator with workflow parameters."""
self.wf_params = wf_params self.wf_params = wf_params
self.options = options or {}
self.alias_field = "sample_name" if self.options.get("no_barcode") else "alias"
self.errors = [] self.errors = []
def log_error(self, msg, column=None, lineno=None): def log_error(self, msg, column=None, lineno=None):
@ -46,13 +48,13 @@ class SampleSheetValidator(ABC):
class ProhibitedColumns(SampleSheetValidator): class ProhibitedColumns(SampleSheetValidator):
"""Check for prohibited columns.""" """Check for prohibited columns."""
def __init__(self, wf_params): def __init__(self, wf_params, options=None):
"""Initialize with workflow parameters.""" """Initialize with workflow parameters."""
super().__init__(wf_params) super().__init__(wf_params, options)
def on_header(self, header): def on_header(self, header):
"""Don't allow `barcode` and `alias` when `--no_barcode` set.""" """Don't allow `barcode` and `alias` when `--no_barcode` set."""
if self.wf_params.get("no_barcode"): if self.options.get("no_barcode"):
for field in ["alias", "barcode"]: for field in ["alias", "barcode"]:
if field in header: if field in header:
self.log_error( self.log_error(
@ -68,9 +70,10 @@ class RequiredColumns(SampleSheetValidator):
"""Check for required columns.""" """Check for required columns."""
self.header = header self.header = header
required = ["barcode", "alias"] required = [self.alias_field]
if self.wf_params.get("no_barcode"): if not self.options.get("no_barcode"):
required = ['sample_name'] # in barcode mode required are alias_field and barcode
required.insert(0, "barcode")
for req in required: for req in required:
if req not in header: if req not in header:
@ -89,9 +92,9 @@ class SampleTypeRules(SampleSheetValidator):
"test_sample", "positive_control", "negative_control", "no_template_control" "test_sample", "positive_control", "negative_control", "no_template_control"
} }
def __init__(self, wf_params): def __init__(self, wf_params, options=None):
"""Initialize with workflow parameters.""" """Initialize with workflow parameters."""
super().__init__(wf_params) super().__init__(wf_params, options)
self.types = [] self.types = []
self.unexpected_types = [] self.unexpected_types = []
self.required = wf_params.get("required_sample_types", []) self.required = wf_params.get("required_sample_types", [])
@ -123,38 +126,18 @@ class SampleTypeRules(SampleSheetValidator):
return valid return valid
class NonEmptySampleSheet(SampleSheetValidator):
"""Check that the sample sheet has at least one data row."""
def __init__(self, wf_params):
"""Initialize with workflow parameters."""
super().__init__(wf_params)
self.row_count = 0
def add_sheet_row(self, row, lineno):
"""Count sample rows."""
self.row_count += 1
@property
def is_valid(self):
"""Check the sample sheet has at least one data row."""
if self.row_count == 0 and not self.errors:
self.log_error("Sample sheet must contain at least one data row")
return super().is_valid
class BarcodeRules(SampleSheetValidator): class BarcodeRules(SampleSheetValidator):
"""Check barcode rules.""" """Check barcode rules."""
def __init__(self, wf_params): def __init__(self, wf_params, options=None):
"""Initialize with workflow parameters.""" """Initialize with workflow parameters."""
super().__init__(wf_params) super().__init__(wf_params, options)
self.first_len = None self.first_len = None
self.barcodes = [] self.barcodes = []
def add_sheet_row(self, row, lineno): def add_sheet_row(self, row, lineno):
"""Check barcode rules.""" """Check barcode rules."""
if self.wf_params.get("no_barcode"): if self.options.get("no_barcode"):
return return
bc = row.get("barcode") bc = row.get("barcode")
@ -185,45 +168,44 @@ class BarcodeRules(SampleSheetValidator):
class AliasRules(SampleSheetValidator): class AliasRules(SampleSheetValidator):
"""Alias rules.""" """Alias rules."""
ALIAS_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$") ALIAS_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
def __init__(self, wf_params): def __init__(self, wf_params, options=None):
"""Initialize with workflow parameters.""" """Initialize with workflow parameters."""
super().__init__(wf_params) super().__init__(wf_params, options)
self.aliases = set() self.aliases = set()
def add_sheet_row(self, row, lineno): def add_sheet_row(self, row, lineno):
"""Check alias rules.""" """Check alias rules."""
if self.wf_params.get("no_barcode"): alias = row.get(self.alias_field)
return
alias = row.get("alias")
if alias is None: if alias is None:
self.log_error("Column missing", column="alias") self.log_error("Column missing", column=self.alias_field)
return return
if alias in self.aliases: if alias in self.aliases:
self.log_error( self.log_error(
f"Value not unique: {alias}", column="alias", lineno=lineno f"Value not unique: {alias}", column=self.alias_field, lineno=lineno,
) )
self.aliases.add(alias) self.aliases.add(alias)
# Specific error message if empty "" to improve user error message
if not alias: if not alias:
self.log_error( self.log_error(
f"Empty value: {alias}. Allowed values start with letters or numbers " f"Empty {self.alias_field}. "
"Allowed values start with letters or numbers "
"and may contain only letters, numbers, '.', '_' or '-'", "and may contain only letters, numbers, '.', '_' or '-'",
column="alias", column=self.alias_field,
lineno=lineno, lineno=lineno,
) )
if not self.ALIAS_PATTERN.match(alias): if not self.ALIAS_PATTERN.match(alias):
self.log_error( self.log_error(
f"Invalid value {alias}. Allowed values start with letters or numbers " f"Invalid value {alias}. Allowed values start with letters or numbers "
"and may contain only letters, numbers, '.', '_' or '-'", "and may contain only letters, numbers, '.', '_' or '-'",
column="alias", column=self.alias_field,
lineno=lineno, lineno=lineno,
) )
if alias.startswith("barcode"): if alias.startswith("barcode"):
self.log_error( self.log_error(
f"Value must not begin with 'barcode': {alias}", f"Value must not begin with 'barcode': {alias}",
column="alias", column=self.alias_field,
lineno=lineno, lineno=lineno,
) )
@ -231,9 +213,9 @@ class AliasRules(SampleSheetValidator):
class AnalysisGroupCompleteness(SampleSheetValidator): class AnalysisGroupCompleteness(SampleSheetValidator):
"""Analysis groups.""" """Analysis groups."""
def __init__(self, wf_params): def __init__(self, wf_params, options=None):
"""Initialize with workflow parameters.""" """Initialize with workflow parameters."""
super().__init__(wf_params) super().__init__(wf_params, options)
self.has_analysis_groups = False self.has_analysis_groups = False
def on_header(self, header): def on_header(self, header):