194 lines
7.1 KiB
Python
194 lines
7.1 KiB
Python
"""Check if a sample sheet is valid."""
|
|
import codecs
|
|
import csv
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
from ..util import get_named_logger, wf_parser # noqa: ABS101
|
|
|
|
|
|
# Some Excel users save their CSV as UTF-8 (and occasionally for a reason beyond my
|
|
# comprehension, UTF-16); Excel then adds a byte order mark (unnecessarily for UTF-8
|
|
# I should add). If we do not handle this with the correct encoding, the mark will
|
|
# appear in the parsed data, causing the header to be malformed.
|
|
# See CW-2310
|
|
def determine_codec(f):
|
|
"""Peek at a file and return an appropriate reading codec."""
|
|
with open(f, 'rb') as f_bytes:
|
|
# Could use chardet here if we need to expand codec support
|
|
initial_bytes = f_bytes.read(8)
|
|
|
|
for codec, encoding_name in [
|
|
[codecs.BOM_UTF8, "utf-8-sig"], # use the -sig codec to drop the mark
|
|
[codecs.BOM_UTF16_BE, "utf-16"], # don't specify LE or BE to drop mark
|
|
[codecs.BOM_UTF16_LE, "utf-16"],
|
|
[codecs.BOM_UTF32_BE, "utf-32"], # handle 32 for completeness
|
|
[codecs.BOM_UTF32_LE, "utf-32"], # again skip LE or BE to drop mark
|
|
]:
|
|
if initial_bytes.startswith(codec):
|
|
return encoding_name
|
|
return None # will cause file to be opened with default encoding
|
|
|
|
|
|
def main(args):
|
|
"""Run the entry point."""
|
|
logger = get_named_logger("checkSheet")
|
|
|
|
barcodes = []
|
|
aliases = []
|
|
sample_types = []
|
|
analysis_groups = []
|
|
allowed_sample_types = [
|
|
"test_sample", "positive_control", "negative_control", "no_template_control"
|
|
]
|
|
|
|
if not os.path.exists(args.sample_sheet) or not os.path.isfile(args.sample_sheet):
|
|
sys.stdout.write("Could not open sample sheet file.")
|
|
sys.exit()
|
|
|
|
try:
|
|
encoding = determine_codec(args.sample_sheet)
|
|
with open(args.sample_sheet, "r", encoding=encoding) as f:
|
|
try:
|
|
# Excel files don't throw any error until here
|
|
csv.Sniffer().sniff(f.readline())
|
|
f.seek(0) # return to initial position again
|
|
except Exception as e:
|
|
# Excel fails with UniCode error
|
|
sys.stdout.write(
|
|
"The sample sheet doesn't seem to be a CSV file.\n"
|
|
"The sample sheet has to be a CSV file.\n"
|
|
"Please verify that the sample sheet is a CSV file.\n"
|
|
f"Parsing error: {e}"
|
|
)
|
|
|
|
sys.exit()
|
|
csv_reader = csv.DictReader(f)
|
|
columns = csv_reader.fieldnames
|
|
alias_field = "alias"
|
|
required_fields = ["barcode"]
|
|
prohibited_fields = []
|
|
|
|
if args.no_barcode:
|
|
alias_field = "sample_name"
|
|
required_fields = []
|
|
prohibited_fields = ["alias", "barcode"]
|
|
|
|
required_fields.append(alias_field)
|
|
|
|
for field in prohibited_fields:
|
|
if field in columns:
|
|
sys.stdout.write(
|
|
f"'{field}' column must not be present with --no_barcode"
|
|
)
|
|
sys.exit()
|
|
|
|
for field in required_fields:
|
|
if field not in columns:
|
|
sys.stdout.write(f"'{field}' column missing")
|
|
sys.exit()
|
|
|
|
# Skip header row for n_row
|
|
for n_row, row in enumerate(csv_reader, start=1):
|
|
if len(row) != len(columns):
|
|
sys.stdout.write(
|
|
f"Unexpected number of cells in row number {n_row}"
|
|
)
|
|
sys.exit()
|
|
if not args.no_barcode:
|
|
barcodes.append(row.get("barcode"))
|
|
aliases.append(row.get(alias_field))
|
|
|
|
# Optional fields check for not None, empty strings are falsey
|
|
sample_type = row.get("type")
|
|
if sample_type is not None:
|
|
sample_types.append(sample_type)
|
|
analysis_group = row.get("analysis_group")
|
|
if analysis_group is not None:
|
|
analysis_groups.append(analysis_group)
|
|
except Exception as e:
|
|
sys.stdout.write(f"Parsing error: {e}")
|
|
sys.exit()
|
|
|
|
# check barcodes are correct format
|
|
for barcode in barcodes:
|
|
if not re.match(r'^barcode\d\d+$', barcode):
|
|
sys.stdout.write("values in 'barcode' column are incorrect format")
|
|
sys.exit()
|
|
|
|
# check aliases are correct format
|
|
# for now we have decided they may not start with "barcode"
|
|
for alias in aliases:
|
|
if alias.startswith("barcode"):
|
|
sys.stdout.write(
|
|
f"values in '{alias_field}' column must "
|
|
"not begin with 'barcode'")
|
|
sys.exit()
|
|
|
|
# check barcodes are all the same length
|
|
if barcodes:
|
|
first_length = len(barcodes[0])
|
|
for barcode in barcodes[1:]:
|
|
if len(barcode) != first_length:
|
|
sys.stdout.write("values in 'barcode' column are different lengths")
|
|
sys.exit()
|
|
|
|
# check barcode and alias values are unique
|
|
if len(barcodes) > len(set(barcodes)):
|
|
sys.stdout.write("values in 'barcode' column not unique")
|
|
sys.exit()
|
|
if len(aliases) > len(set(aliases)):
|
|
sys.stdout.write(f"values in '{alias_field}' column not unique")
|
|
sys.exit()
|
|
|
|
if sample_types:
|
|
# check if "type" column has unexpected values
|
|
unexp_type_vals = set(sample_types) - set(allowed_sample_types)
|
|
|
|
if unexp_type_vals:
|
|
sys.stdout.write(
|
|
f"found unexpected values in 'type' column: {unexp_type_vals}. "
|
|
f"Allowed values are: {allowed_sample_types}"
|
|
)
|
|
sys.exit()
|
|
|
|
if args.required_sample_types:
|
|
for required_type in args.required_sample_types:
|
|
if required_type not in allowed_sample_types:
|
|
sys.stdout.write(f"Not an allowed sample type: {required_type}")
|
|
sys.exit()
|
|
if sample_types.count(required_type) < 1:
|
|
sys.stdout.write(
|
|
f"Sample sheet requires at least 1 of {required_type}")
|
|
sys.exit()
|
|
if analysis_groups:
|
|
# if there was a "analysis_group" column, make sure it had values for all
|
|
# samples
|
|
if not all(analysis_groups):
|
|
sys.stdout.write(
|
|
"if an 'analysis_group' column exists, it needs values in each row"
|
|
)
|
|
sys.exit()
|
|
|
|
logger.info(f"Checked sample sheet {args.sample_sheet}.")
|
|
|
|
|
|
def argparser():
|
|
"""Argument parser for entrypoint."""
|
|
parser = wf_parser("check_sample_sheet")
|
|
parser.add_argument("sample_sheet", help="Sample sheet to check")
|
|
parser.add_argument(
|
|
"--required_sample_types",
|
|
help="List of required sample types. Each sample type provided must "
|
|
"appear at least once in the sample sheet",
|
|
nargs="*"
|
|
)
|
|
parser.add_argument(
|
|
"--no_barcode",
|
|
action="store_true",
|
|
help="Allow sample sheets without a barcode column "
|
|
"and match rows by sample_name only",
|
|
)
|
|
return parser
|