Consider replacing SAMPLESHEET_CHECK module with the checks in schema_input.json

[Vivian0105]

Description of feature

The samplesheet check (SAMPLESHEET_CHECK) might have been replaced by the checks in schema_input.json

[stela2502]

I found the sample check incredible unhelpful for a lab scientists sample sheet.
The error messages are incredibly cryptic - typical Python error messages that omit just the useful information and leave you clueless of what the real issues actually are.

A ChatGPT replacement for your sample check would be this:

#!/usr/bin/env python

import os
import sys
import errno
import argparse
import pandas as pd
import re


def parse_args(args=None):
    parser = argparse.ArgumentParser(
        description="Read nf-core/airrflow samplesheet file and check its contents.",
        epilog="Example usage: python check_samplesheet.py <FILE_IN>",
    )
    parser.add_argument("file_in", help="Input samplesheet file.")
    parser.add_argument(
        "-a", "--assembled",
        help="Input samplesheet type",
        action="store_true",
        default=False,
    )
    return parser.parse_args(args)


def fail(msg):
    print(f"\nERROR: Please check input samplesheet\n{msg}\n", file=sys.stderr)
    sys.exit(1)


def warn(msg):
    print(f"\nWARNING: {msg}\n", file=sys.stderr)


def clean_cell(x):
    if pd.isna(x):
        return ""
    return str(x).strip().strip('"')


def check_no_whitespace(tab, col, forbidden_cols):
    for idx, value in tab[col].items():
        value = clean_cell(value)

        if value == "":
            fail(
                f"Missing value in required no-whitespace column.\n"
                f"Row: {idx + 2}\n"
                f"Column: {col}"
            )

        if re.search(r"\s+", value):
            fail(
                f"Whitespace found in column where whitespace is not allowed.\n"
                f"Row: {idx + 2}\n"
                f"Column: {col}\n"
                f"Value: {value!r}\n"
                f"Columns that must not contain whitespace: {', '.join(forbidden_cols)}"
            )


def check_required_columns(header, required_columns):
    missing = [col for col in required_columns if col not in header]

    if missing:
        fail(
            f"Samplesheet header is missing required column(s): {', '.join(missing)}\n"
            f"Detected header:\n{repr(header)}\n"
            f"Required columns:\n{', '.join(required_columns)}"
        )


def check_row_widths(file_in, header, min_cols):
    expected = len(header)

    with open(file_in, "r", encoding="utf-8") as fin:
        next(fin)

        for line_num, line in enumerate(fin, start=2):
            raw = line.rstrip("\n")
            fields = [x.strip().strip('"') for x in raw.split("\t")]

            if len(fields) != expected:
                fail(
                    f"Invalid number of columns.\n"
                    f"Row: {line_num}\n"
                    f"Expected columns: {expected}\n"
                    f"Observed columns: {len(fields)}\n"
                    f"Line:\n{raw!r}\n\n"
                    f"Hint: this usually means the file is not truly tab-separated, "
                    f"or one free-text field contains tabs."
                )

            populated = len([x for x in fields if x != ""])
            if populated < min_cols:
                fail(
                    f"Too few populated columns.\n"
                    f"Row: {line_num}\n"
                    f"Required populated columns at minimum: {min_cols}\n"
                    f"Observed populated columns: {populated}\n"
                    f"Line:\n{raw!r}"
                )


def check_allowed_values(tab, col, allowed, case_insensitive=False):
    allowed_display = ", ".join(allowed)

    for idx, value in tab[col].items():
        value = clean_cell(value)

        cmp_value = value.upper() if case_insensitive else value
        cmp_allowed = [x.upper() for x in allowed] if case_insensitive else allowed

        if cmp_value not in cmp_allowed:
            fail(
                f"Invalid value.\n"
                f"Row: {idx + 2}\n"
                f"Column: {col}\n"
                f"Value: {value!r}\n"
                f"Allowed values: {allowed_display}"
            )


def check_samplesheet(file_in, assembled):
    min_cols = 7

    required_columns_raw = [
        "sample_id",
        "filename_R1",
        "filename_R2",
        "subject_id",
        "species",
        "pcr_target_locus",
        "single_cell",
        "sex",
        "tissue",
        "biomaterial_provider",
        "age",
    ]

    required_columns_assembled = [
        "filename",
        "sample_id",
        "subject_id",
        "species",
        "pcr_target_locus",
        "sex",
        "tissue",
        "biomaterial_provider",
        "age",
        "single_cell",
    ]

    no_whitespaces_raw = [
        "sample_id",
        "filename_R1",
        "filename_R2",
        "subject_id",
        "species",
        "pcr_target_locus",
    ]

    no_whitespaces_assembled = [
        "sample_id",
        "filename",
        "subject_id",
        "species",
        "pcr_target_locus",
    ]

    required_columns = required_columns_assembled if assembled else required_columns_raw
    no_whitespaces = no_whitespaces_assembled if assembled else no_whitespaces_raw

    try:
        with open(file_in, "r", encoding="utf-8") as fin:
            first_line = fin.readline().rstrip("\n")
    except FileNotFoundError:
        fail(f"File does not exist: {file_in}")

    if not first_line:
        fail("Samplesheet is empty.")

    header = [x.strip().strip('"') for x in first_line.split("\t")]

    if len(header) == 1:
        fail(
            "Header contains only one column.\n"
            "This file is probably not tab-separated.\n"
            f"Detected header line:\n{first_line!r}"
        )

    check_required_columns(header, required_columns)
    check_row_widths(file_in, header, min_cols)

    try:
        tab = pd.read_csv(
            file_in,
            sep="\t",
            header=0,
            dtype=str,
            keep_default_na=False,
            na_filter=False,
        )
    except Exception as e:
        fail(f"Could not read samplesheet with pandas as TSV.\nReason: {e}")

    tab.columns = [clean_cell(c) for c in tab.columns]

    for col in required_columns:
        if col not in tab.columns:
            fail(f"Required column disappeared after pandas parsing: {col}")

        for idx, value in tab[col].items():
            if clean_cell(value) == "":
                fail(
                    f"Missing required value.\n"
                    f"Row: {idx + 2}\n"
                    f"Column: {col}"
                )

    for col in no_whitespaces:
        check_no_whitespace(tab, col, no_whitespaces)

    if len(tab["sample_id"]) != len(set(tab["sample_id"])):
        duplicated = tab.loc[tab["sample_id"].duplicated(), "sample_id"].tolist()

        if assembled:
            fail(
                "Sample IDs are not unique.\n"
                f"Duplicated sample_id value(s): {duplicated}"
            )
        else:
            warn(
                "Sample IDs are not unique. FastQs with the same sample ID will be merged.\n"
                f"Duplicated sample_id value(s): {duplicated}"
            )

    check_allowed_values(
        tab,
        "pcr_target_locus",
        ["IG", "TR"],
        case_insensitive=True,
    )

    check_allowed_values(
        tab,
        "species",
        ["human", "mouse"],
        case_insensitive=False,
    )

    for subject_id, subtab in tab.groupby("subject_id", dropna=False):
        species_values = sorted(set(clean_cell(x) for x in subtab["species"]))

        if len(species_values) > 1:
            fail(
                "The same subject_id cannot belong to different species.\n"
                f"subject_id: {subject_id!r}\n"
                f"Observed species values: {species_values}"
            )

    print("Samplesheet validation passed.")


def main(args=None):
    args = parse_args(args)
    check_samplesheet(args.file_in, args.assembled)


if __name__ == "__main__":
    sys.exit(main())

But I assume this does not create a validated file. I have not checked that in regards to your pipeline, but a file that passed this check did then also work in your pipeline.

Would be absolutely cool if you could use the improved error messages!

And it actually took me 2 min to get that...