From f702b6e9e02d374fea2b8ae13a901020621d9cf2 Mon Sep 17 00:00:00 2001 From: Charles Monod-Broca Date: Thu, 12 Feb 2026 11:21:48 +0100 Subject: [PATCH] Added singularity def file to ease deployment Also got rid of many dependencies --- .gitignore | 1 + README.md | 31 +- app.py => app/app.py | 52 +- app/checker.py | 58 ++ .../examples}/Bad_SampleSheet_alphanum.csv | 0 .../Bad_SampleSheet_extra_semicolons.csv | 0 {examples => app/examples}/sample_sheet.csv | 0 .../examples}/sample_sheet_settings_index.csv | 0 {imgs => app/imgs}/logo_256x256.png | Bin requirements.txt => app/requirements.txt | 2 +- app/samplesheet.py | 944 ++++++++++++++++++ samplesheet_validator.def | 12 + 12 files changed, 1078 insertions(+), 22 deletions(-) create mode 100644 .gitignore rename app.py => app/app.py (90%) create mode 100644 app/checker.py rename {examples => app/examples}/Bad_SampleSheet_alphanum.csv (100%) rename {examples => app/examples}/Bad_SampleSheet_extra_semicolons.csv (100%) rename {examples => app/examples}/sample_sheet.csv (100%) rename {examples => app/examples}/sample_sheet_settings_index.csv (100%) rename {imgs => app/imgs}/logo_256x256.png (100%) rename requirements.txt => app/requirements.txt (57%) create mode 100644 app/samplesheet.py create mode 100644 samplesheet_validator.def diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed8ebf5 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__ \ No newline at end of file diff --git a/README.md b/README.md index ef7e2bd..cea5b68 100644 --- a/README.md +++ b/README.md @@ -15,13 +15,32 @@ The sanity checks implemented are based on experience and the bcl2fastq document # Local instance - git clone https://github.com/sequana/webapp_samplesheet check_my_sample_sheet - cd check_my_sample_sheet +## Using local python - # You will need to install requirements (sequana and streamlit) - pip install --file requirements.txt +If you'd like to run a local instance, you could either install streamlit and run it directly (python virtual environment is highly recommended): - # and should ne ready to test the appliction locally in your browser - streamlit run app.py +```bash +git clone https://github.com/sequana/webapp_samplesheet check_my_sample_sheet +cd check_my_sample_sheet/app +pip install -r requirements.txt +python -m streamlit run app.py +``` +## Using singularity +Or if you'd like to run an instance on a web server, there is a singularity definition file (requires singularity and docker to build, only singularity to run): + +```bash +git clone https://github.com/sequana/webapp_samplesheet check_my_sample_sheet +singularity build --build-arg SERVER_ADDRESS=127.0.0.1 --fakeroot sample_sheet_validator_1.0.0.sif samplesheet_validator.def +``` + +You can change the IP address you're listening to to suit your needs, and then run an instance of the app: + +```bash +singularity instance start sample_sheet_validator_1.0.0.sif samplesheet_validatord +``` + +Please note: if using singularity, make sure that both `/tmp` and `$HOME` are writable. You can either set `mount tmp = yes` and `mount home = yes` in `/etc/singularity/singularity.conf`, or you can set the binds manually. + +By default, the app will be served on port 8501, so if you'd like to redirect http traffic through a reverse proxy, you can specify the IP address and port 8501 as the target. diff --git a/app.py b/app/app.py similarity index 90% rename from app.py rename to app/app.py index dd20b28..d3d8560 100644 --- a/app.py +++ b/app/app.py @@ -17,14 +17,16 @@ import requests import streamlit as st -from sequana.iem import SampleSheet +from samplesheet import SampleSheet from streamlit_option_menu import option_menu st.set_page_config( page_title="Illumina Sample Sheet Validator", page_icon="imgs/logo_256x256.png", layout="wide", - menu_items={"Report a bug": "https://github.com/sequana/webapp_samplesheet/issues/new/choose"}, + menu_items={ + "Report a bug": "https://github.com/sequana/webapp_samplesheet/issues/new/choose" + }, ) version = "1.0.0" @@ -104,7 +106,9 @@ def add_legend(success, warning, error): error = counter["Error"] / S * 100 completed = min(round(100 * (S / float(N))), 100) - bar_placeholder.markdown(colored_bar(success, warning, error, completed), unsafe_allow_html=True) + bar_placeholder.markdown( + colored_bar(success, warning, error, completed), unsafe_allow_html=True + ) # finally add the legend _, col2, _ = st.columns([1, 4, 1]) @@ -126,7 +130,9 @@ def add_legend(success, warning, error): def main(): - st.sidebar.write("Provided by the [Sequana teams](https://github.com/sequana/sequana)") + st.sidebar.write( + "Provided by the [Sequana teams](https://github.com/sequana/sequana)" + ) st.sidebar.image("imgs/logo_256x256.png") st.title(f"Sample Sheet and Design Validator (v{version})") @@ -135,7 +141,11 @@ def main(): # 1. as sidebar menu with st.sidebar: choice = option_menu( - "Main Menu", menu, icons=["gear", "gear", "cloud-upload", ""], menu_icon="cast", default_index=0 + "Main Menu", + menu, + icons=["gear", "gear", "cloud-upload", ""], + menu_icon="cast", + default_index=0, ) if choice == "Sample Sheet Validation (Illumina)": @@ -149,15 +159,21 @@ def main(): col1, col2, col3 = st.columns([4, 1, 4]) with col1: data_file = st.file_uploader( - "Drop a sample sheet below and press the **Process** button. ", type=["csv", "txt"] + "Drop a sample sheet below and press the **Process** button. ", + type=["csv", "txt"], ) with col2: # Centered "OR" text - st.markdown("


OR
", unsafe_allow_html=True) + st.markdown( + "


OR
", + unsafe_allow_html=True, + ) with col3: code = st.text_area( - "Paste your code here and press the **Process** button.", value=st.session_state.code, key="code_area" + "Paste your code here and press the **Process** button.", + value=st.session_state.code, + key="code_area", ) if st.button(":gear: Process :gear:"): @@ -216,9 +232,7 @@ def main(): st.code(data, language="bash") st.subheader("3 - [Data] section with single-index and a [Settings] section") - url = ( - "https://raw.githubusercontent.com/sequana/webapp_samplesheet/main/examples/sample_sheet_settings_index.csv" - ) + url = "https://raw.githubusercontent.com/sequana/webapp_samplesheet/main/examples/sample_sheet_settings_index.csv" r = requests.get(url, allow_redirects=True) data = r.content.decode() st.code(data, language="bash") @@ -229,7 +243,9 @@ def main(): data = r.content.decode() st.code(data, language="bash") - st.subheader("5 - Example of an erroneous sample sheet (extra semicolons at the end)") + st.subheader( + "5 - Example of an erroneous sample sheet (extra semicolons at the end)" + ) url = "https://raw.githubusercontent.com/sequana/webapp_samplesheet/main/examples/Bad_SampleSheet_extra_semicolons.csv" r = requests.get(url, allow_redirects=True) data = r.content.decode() @@ -249,7 +265,9 @@ def main(): ) else: st.subheader("How to cite ? ") - st.info("Check My Sample Sheet application (this website):\n\nLemée L. et al [researchsquare](https://www.researchsquare.com/article/rs-5268893/v1)") + st.info( + "Check My Sample Sheet application (this website):\n\nLemée L. et al [researchsquare](https://www.researchsquare.com/article/rs-5268893/v1)" + ) st.info( "The Sequana application to check the sample sheet: \n\nCokelaer T. et al, (2017), 'Sequana': a Set of Snakemake NGS pipelines, Journal of Open Source Software, 2(16), 352, JOSS DOI [doi:10.21105/joss.00352](https://joss.theoj.org/papers/10.21105/joss.00352)" @@ -272,11 +290,15 @@ def process_sample_sheet(data_file, samplesheet): None """ if data_file is not None: - file_details = {"Filename": data_file.name, "FileType": data_file.type, "FileSize": data_file.size} + file_details = { + "Filename": data_file.name, + "FileType": data_file.type, + "FileSize": data_file.size, + } else: pass - if 1 == 1: + if True: # read to save locally # samplesheet = data_file.read().decode() diff --git a/app/checker.py b/app/checker.py new file mode 100644 index 0000000..11c11ba --- /dev/null +++ b/app/checker.py @@ -0,0 +1,58 @@ +# +# This file is part of Sequana software +# +# Copyright (c) 2018-2022 - Sequana Development Team +# +# Distributed under the terms of the 3-clause BSD license. +# The full license is in the LICENSE file, distributed with this software. +# +# website: https://github.com/sequana/sequana +# documentation: http://sequana.readthedocs.io +# +############################################################################## + +import colorlog + +logger = colorlog.getLogger(__name__) + + +class Checker: + """Utility to hold checks + + + The method :meth:`~sequana.utils.checks.Checke./tryme` calls the method or function + provided. This method is expected to return a dictionary + with 2 keys called status and msg. Status should be + in 'Error', 'Warning', 'Success'. + + The attributes hold all calls to :meth:`tryme` + + Given that func returns a dictionary as explained here above, you can run this code + :: + + checks = Checker() + checks.tryme(func) + + checks contains the status and mesg of each function called by checks.tryme. + + + """ + + def __init__(self): + + self.results = [] + + def tryme(self, meth): + try: + status = meth() + if "msg" in status and "status" in status: + self.results.append(status) + else: + self.results.append({"status": "Success", "msg": status}) + except Exception as err: # pragma: no cover + self.results.append( + {"msg": err, "status": "Error", "caller": str(meth.__name__)} + ) + self.results.append( + {"msg": err, "status": "Error", "caller": str(meth.__name__)} + ) diff --git a/examples/Bad_SampleSheet_alphanum.csv b/app/examples/Bad_SampleSheet_alphanum.csv similarity index 100% rename from examples/Bad_SampleSheet_alphanum.csv rename to app/examples/Bad_SampleSheet_alphanum.csv diff --git a/examples/Bad_SampleSheet_extra_semicolons.csv b/app/examples/Bad_SampleSheet_extra_semicolons.csv similarity index 100% rename from examples/Bad_SampleSheet_extra_semicolons.csv rename to app/examples/Bad_SampleSheet_extra_semicolons.csv diff --git a/examples/sample_sheet.csv b/app/examples/sample_sheet.csv similarity index 100% rename from examples/sample_sheet.csv rename to app/examples/sample_sheet.csv diff --git a/examples/sample_sheet_settings_index.csv b/app/examples/sample_sheet_settings_index.csv similarity index 100% rename from examples/sample_sheet_settings_index.csv rename to app/examples/sample_sheet_settings_index.csv diff --git a/imgs/logo_256x256.png b/app/imgs/logo_256x256.png similarity index 100% rename from imgs/logo_256x256.png rename to app/imgs/logo_256x256.png diff --git a/requirements.txt b/app/requirements.txt similarity index 57% rename from requirements.txt rename to app/requirements.txt index 6dfc758..20b3f63 100644 --- a/requirements.txt +++ b/app/requirements.txt @@ -1,2 +1,2 @@ -sequana==0.17.1 streamlit_option_menu +colorlog diff --git a/app/samplesheet.py b/app/samplesheet.py new file mode 100644 index 0000000..db2f30f --- /dev/null +++ b/app/samplesheet.py @@ -0,0 +1,944 @@ +# +# This file is part of Sequana software +# +# Copyright (c) 2018-2022 - Sequana Development Team +# +# Distributed under the terms of the 3-clause BSD license. +# The full license is in the LICENSE file, distributed with this software. +# +# website: https://github.com/sequana/sequana +# documentation: http://sequana.readthedocs.io +# +############################################################################## +"IEM class" +import io +import os +import sys + +import colorlog +import pandas as pd + +logger = colorlog.getLogger(__name__) + + +__all__ = ["SampleSheet", "IEM"] + + +class SampleSheet: + """Reader and validator of Illumina samplesheets + + The Illumina samplesheet reader and validator verifies the correctness of the sections + in the samplesheet, which are not case-sensitive and are enclosed within square brackets. + + Following the closing bracket, no additional characters are permitted, except + for commas and the end-of-line marker. For instance [Data]a prevents the [Data] section + from being correctly processed. + + The sections then consist of key-value pairs represented as records, with each line consisting + of precisely two fields. + + An optional [Settings] section can contain key-value pairs, and + the [Reads] section specifies the number of cycles per read, which + is exclusively required for MiSeq. + + The [Data] section, which is a table similar to CSV format, is optional. + However, without [Data] section all reads are sent to a single 'undetermined' + output file. Sample_ID is highly recommended. + + Example of typical Data section to be used with bcl2fastq:: + + [Header] + + [Data] + Sample_ID,Sample_Name,I7_Index_ID,index,I5_INdex_ID,index2 + A10001,Sample_A,D701,AATACTCG,D501,TATAGCCT + A10002,Sample_B,D702,TCCGGAGA,D501,TATAGCCT + A10003,Sample_C,D703,CGCTCATT,D501,TATAGCCT + A10004,Sample_D,D704,GAGATTCC,D501,TATAGCCT + + Important: altough we have upper case names as specified in the Illumina specs, the + bcl2fastq does not care about the upper case. This is not intuitive since IEM produces + keys with upper and lower case names similarly to the specs. + + **Sequana Standalone** + + The standalone application **sequana** contains a subcommand based on this class:: + + sequana samplesheet + + that can be used to check the correctness of a samplesheet:: + + sequana samplesheet --check SampleSheet.csv + + :references: illumina specifications 970-2017-004.pdf + """ + + expected_headers_fields = [ + "IEMFileVersion", + "Investigator Name", + "Instrument Type", + "Experiment Name", + "Date", + "Workflow", + "Application", + "Assay", + "Description", + "Chemistry", + "Index Adapters", + ] + + expected_data_headers = {"SE": [], "PE": []} + + def __init__(self, filename): + + self.filename = filename + if not os.path.exists(self.filename): + raise IOError(f"{filename} does not exist") + # figures out the sections in the sample sheet. + # we use a try/except so that even in case of failure, we can still use + # quickfix or attributes. + try: + self._scan_sections() + except Exception as err: # pragma: no cover + print(err) + + def _line_cleaner(self, line, line_count): + # We can get rid of EOL and spaces + line = line.strip() + + # is it an empty line ? + if len(line) == 0: + return line + + # if we are dealing with a section title, we can cleanup the + # line. A section must start with '[' and ends with ']' but + # there could be spaces and commands. If it ends with a ; then + # the section will not be found as expected since this is + # sympatomatic of further issues + if line.startswith("["): + # [Header], ,, ,\n becomes [Header] + line = line.strip(", ;") # note the space AND comma + + return line + + def _scan_sections(self): + # looks for special section Header/Data/Reads/Settings or + # any other matching section thst looks like [XXX] + + # sections can be of any types of cases (lower, upper, mixed) + + current_section = None + data = {} + with open(self.filename, "r") as fin: + for line_count, line in enumerate(fin.readlines()): + line = self._line_cleaner(line, line_count + 1) + if len(line) == 0: + continue + if line.startswith("[") and line.endswith("]"): + name = line.lstrip("[").rstrip("]") + current_section = name + data[current_section] = ( + [] + ) # create empty list just to create the section + else: + if current_section in data: + data[current_section] += [line] + else: + data[current_section] = [line] + self.sections = data + + # if the sample sheet starts with an incorrect name + # e.g section not closed by square brackets or empty lines + # the a None key is present and should be ignored or cast + # into a string: + if None in self.sections: + self.sections["None"] = self.sections[None] + del self.sections[None] + + # Some cleanup. Since the sections are case insensitive within + # bcl2fastq, we need to convert everything back to titles + self.sections = {k.title(): v for k, v in self.sections.items()} + + def _get_df(self): + if "Data" in self.sections: + if self.sections["Data"]: + # cope with the case of comma or semicolon separators. + df1 = pd.read_csv( + io.StringIO("\n".join(self.sections["Data"])), + index_col=False, + sep=",", + ) + df2 = pd.read_csv( + io.StringIO("\n".join(self.sections["Data"])), + index_col=False, + sep=";", + ) + + if len(df1.columns) > len(df2.columns): + + # rename all columns to lower case + df1.columns = [x.lower() for x in df1.columns] + return df1 + else: + df2.columns = [x.lower() for x in df2.columns] + return df2 + else: + return pd.DataFrame() + else: # pragma: no cover + return pd.DataFrame() + + df = property(_get_df, doc="Returns the [Data] section") + + def _get_samples(self): + try: + return self.df["sample_id"].values + except AttributeError: # pragma: no cover + return "No Sample_ID found in the Data header section" + + samples = property(_get_samples, doc="returns the sample identifiers as a list") + + def _get_version(self): + try: + return self.header["IEMFileVersion"] + except KeyError: + return None + + version = property(_get_version, doc="return the version of the IEM file") + + def checker(self): + + from checker import Checker + + checks = Checker() + + if "Reads" in self.sections: + pass + else: + checks.results.append( + {"msg": f"The optional [Reads] section is missing", "status": "Warning"} + ) + + if "Header" in self.sections: + pass + else: + checks.results.append( + { + "msg": f"The optional [Header] section is missing", + "status": "Warning", + } + ) + + if "Data" in self.sections: + checks.tryme(self._check_data_section_csv_format) + checks.tryme(self._check_optional_data_column_names) + checks.tryme(self._check_mandatory_data_columns) + checks.tryme(self._check_sample_ID) + checks.tryme(self._check_sample_project) + checks.tryme(self._check_unique_sample_ID) + checks.tryme(self._check_unique_indices) + checks.tryme(self._check_nucleotide_indices) + checks.tryme(self._check_sample_lane_number) + checks.tryme(self._check_alpha_numerical) + + if "sample_name" in self.df.columns: + checks.tryme(self._check_sample_names) + checks.tryme(self._check_unique_sample_name) + else: + checks.results.append( + { + "name": "check_sample_names", + "msg": f"Column Sample_Name not found in the header of the [Data] section. Recommended.", + "status": "Warning", + } + ) + + # if data section is incorrect, self.df is not accessible so the following validation + # will fail. + try: + if "index" in self.df.columns: + checks.tryme(self._check_homogene_I7_length) + + if "index2" in self.df.columns: + checks.tryme(self._check_homogene_I5_length) + checks.tryme(self._check_homogene_I5_and_I7_length) + except pd.errors.ParserError: # pragma: no cover + pass + else: + checks.results.append( + {"msg": f"The [Data] section is missing. ", "status": "Error"} + ) + + if "Settings" in self.sections: + checks.tryme(self._check_settings) + else: + checks.results.append( + { + "msg": f"The optional [Settings] section is missing", + "status": "Warning", + } + ) + + checks.tryme(self._check_semi_column_presence) + + return checks.results + + def _check_settings(self): + + for k, v in self.settings.items(): + + # checks ACGT content (no acgt allowed) + if k.lower() in [ + x.lower() + for x in [ + "Adapter", + "TrimAdapter", + "AdapterRead2", + "TrimAdapterRead2", + "MaskAdapter", + "MaskAdapterRead2", + ] + ]: + allowed_chars = set("ACGT") + + def is_valid_string(s): + return set(s).issubset(allowed_chars) + + if is_valid_string(v) is False: + return { + "name": "check_settings", + "msg": f"Invalid nucleotide sequence found for {k} (v)", + "status": "Error", + } + elif k.lower() in [ + x.lower() + for x in [ + "ReverseComplement", + "FindAdaptersWithIndels", + "TrimUMI", + "CreateFastqForIndexReads", + ] + ]: + if v not in ( + "true", + "false", + "t", + "f", + "yes", + "no", + "y", + "n", + "1", + "0", + ): + return { + "name": "check_settings", + "msg": f"Invalid valid for {k} ({v}). Must be set to one of : true, false, t, f, yes, no, y, n, 1, 0", + "status": "Error", + } + + elif k.lower() in [ + x.lower() + for x in [ + "Read1StartFromCycle", + "Read2StartFromCycle", + "Read1EndWithCycle", + "Read2EndWithCycle", + "Read1UMILength", + "Read2UMILength", + "Read1UMIStartFromCycle", + "Read2UMIStartFromCycle", + ] + ]: + allowed_chars = set("0123456789") + + def is_valid_int(s): + return set(s).issubset(allowed_chars) + + if is_valid_int(v) is False or int(v) < 0: + return { + "name": "check_settings", + "msg": f"Invalid value for {k}. Must be positive. You provided: {v}", + "status": "Error", + } + elif k.lower() in [ + x.lower() for x in ["ExcludeTiles", "ExcludeTilesLaneX"] + ]: + + allowed_chars = set("0123456789") + + def is_valid_int(s): + return set(s).issubset(allowed_chars) + + # valid is 1101+1102+1103-1110 (only +-,numbers) + for item in v.split("+"): + for x in item.split("-"): + if is_valid_int(x) is False: + return { + "name": "check_settings", + "msg": f"Invalid value for {k} (v). Must be made of integers, + and - signs. e.g. 1101+1105-1110 to exclude 1101 and values in [1105-1110].", + "status": "Error", + } + else: + return { + "name": "check_settings", + "msg": f"The [Settings] section looks good", + "status": "Success", + } + + def _check_sample_ID(self): + + if "sample_id" in self.df.columns: # optional + # check that names are not in 'all' or 'undetermined' + if (self.df["sample_id"] == "unknown").sum() or ( + self.df["sample_id"] == "all" + ).sum(): + return { + "name": "check_sample_ID", + "msg": "Sample_ID column contains forbidden name ('all' or 'unknown')", + "status": "Error", + } + else: + return { + "name": "check_sample_ID", + "msg": "Sample_ID column (no unknown/undetermined label). Looks correct", + "status": "Success", + } + + else: + return { + "name": "check_sample_ID", + "msg": f"Column Sample_ID not found in the header of the [Data] section. All data will be stored in Undetermined.fastq.gz", + "status": "Warning", + } + + def _check_sample_names(self): + + # check that names are not in 'all' or 'undetermined' + if (self.df["sample_name"] == "unknown").sum() or ( + self.df["sample_name"] == "undetermined" + ).sum(): + return { + "name": "check_sample_names", + "msg": "Sample_Name column contains forbidden name ('all' or 'undetermined')", + "status": "Error", + } + else: + return { + "name": "check_sample_names", + "msg": "Sample_Name column (no unknown/undetermined label). Looks correct", + "status": "Success", + } + + def _check_sample_project(self): + + if "sample_project" in self.df.columns: # optional + # check that names are not in 'all' or 'undetermined' + if (self.df["sample_project"] == "all").sum() or ( + self.df["sample_project"] == "default" + ).sum(): + return { + "name": "check_sample_project", + "msg": "Sample_Project column contains forbidden name ('all' or 'default')", + "status": "Error", + } + else: + return { + "name": "check_sample_project", + "msg": "Sample_Project column (no all/default label). Looks correct", + "status": "Success", + } + else: + return { + "name": "check_sample_project", + "msg": f"Column Sample_Project not found in the header of the [Data] section. Recommended.", + "status": "Warning", + } + + def _check_mandatory_data_columns(self): + # In fact, all columns are optional except index and Sample_Name is optional + # Sample_Project is optional. If provided, fastq are saved in that sub directory. + + for column in ["sample_id", "index"]: + if column not in self.df.columns: + return { + "name": "check_mandatory_data_columns", + "msg": f"Mandatory '{column}' column not found in the header of the [Data] section", + "status": "Error", + } + return { + "name": "check_mandatory_data_columns", + "msg": f"Mandatory columns (index, Sample_ID) found in the header of the [Data] section", + "status": "Success", + } + + def _check_unique_sample_name(self): + # check that sample names are unique and that sample Names are unique too + + if self.df["sample_name"].isnull().sum() > 0: + return { + "name": "check_unique_sample_name", + "msg": "Some sample names are empty", + "status": "Warning", + } + + elif len(self.df.sample_name) != len(self.df.sample_name.unique()): + duplicated = self.df.sample_name[self.df.sample_name.duplicated()].index + duplicated = ",".join([str(x + 1) for x in duplicated]) + return { + "name": "check_unique_sample_name", + "msg": f"Sample_Name not unique. Duplicated entries on lines: {duplicated}", + "status": "Warning", + } + else: + return { + "name": "check_unique_sample_name", + "msg": "Sample name uniqueness", + "status": "Success", + } + + def _check_unique_sample_ID(self): + # check that sample names are unique and that sample Names are unique too + + if "sample_id" not in self.df.columns: + return { + "name": "check_unique_sample_ID", + "msg": "Sample ID not found in the header of the [Data] section", + "status": "Warning", + } + + if len(self.df["sample_id"]) != len(self.df["sample_id"].unique()): + duplicated = self.df.sample_id[self.df.sample_id.duplicated()].index + duplicated = ",".join([str(x + 1) for x in duplicated]) + return { + "name": "check_unique_sample_ID", + "msg": f"Sample ID not unique. Duplicated entries on lines: {duplicated}", + "status": "Error", + } + else: + return { + "name": "check_unique_sample_ID", + "msg": "Sample ID uniqueness", + "status": "Success", + } + + def _check_sample_lane_number(self): + if "sample_lane" in self.df.columns: + + # Define the allowed lanes + allowed_chars = set("12345678") + + def is_valid_lane(s): + return set(str(s)).issubset(allowed_chars) + + # Apply the function to the DataFrame + invalid_lanes = list( + self.df[~self.df["sample_lane"].apply(is_valid_lane)].index + ) + + if len(invalid_lanes): + invalid_lanes = [x + 1 for x in invalid_lanes] + return { + "name": "check_sample_lane_number", + "msg": f"Incorrect lane number in these rows: {invalid_lanes}. Must be in the range 1-8", + "status": "Error", + } + + return { + "name": "check_sample_lane_number", + "msg": "Correct lane number range", + "status": "Success", + } + + def _check_nucleotide_indices(self): + + # Define the allowed characters + allowed_chars = set("ACGTN") + + def is_valid_string(s): + try: + return set(s).issubset(allowed_chars) + except: + return False + + # Apply the function to the DataFrame + invalid_rows = list(self.df[~self.df["index"].apply(is_valid_string)].index) + + if "index2" in self.df.columns: + invalid_rows += list( + self.df[~self.df["index2"].apply(is_valid_string)].index + ) + if len(invalid_rows): + invalid_rows = [x + 1 for x in invalid_rows] + return { + "name": "check_nucleotide_indices", + "msg": f"these rows have invalid index with wrong nucleotides {invalid_rows}", + "status": "Error", + } + + return {"msg": "Indices are made of A, C, G, T, N.", "status": "Success"} + + def _check_unique_indices(self): + if "index2" in self.df.columns: + indices = self.df["index"] + "," + self.df["index2"] + msg = "You have duplicated index I7/I5." + elif "index" in self.df.columns: + indices = self.df["index"] + msg = "You have duplicated index I7." + else: + return { + "name": "check_unique_indices", + "msg": f"column 'index' not found in the header of the [Data] section.", + "status": "Error", + } + + if indices.duplicated().sum() > 0: + duplicated = indices[indices.duplicated()].values + try: + IDs = self.df[indices.duplicated()].sample_id.values + IDs = ", ".join([str(x) for x in IDs]) + IDs = f"related to sample IDs: {IDs}" + except Exception as err: # pragma: no cover + IDs = "" + + return { + "name": "check_unique_indices", + "msg": f"{msg} {duplicated} {IDs}", + "status": "Error", + } + else: + return { + "name": "check_unique_indices", + "msg": "Indices are unique.", + "status": "Success", + } + + def _check_optional_data_column_names(self): + msg = "" + warnings = [] + # check whether minimal columns are included + for x in [ + "i7_index_id", + "sample_project", + "description", + "sample_plate", + "sample_well", + "lane", + "index_plate", + "index_plate_well", + ]: + if x not in self.df.columns: + warnings.append(x) + + if len(warnings): + warnings = ",".join(warnings) + msg = f"Some columns are missing in the [Data] section: {warnings}" + return {"msg": msg, "status": "Warning"} + else: # pragma: no cover + return { + "msg": "Columns of the data section looks good", + "status": "Success", + } + + def _get_data_length(self): + N = len(set([x.count(",") for x in self.sections["Data"]])) + if len(self.sections["Data"]) >= 2 and N == 1: + return True + else: + return False + + def _check_homogene_I7_length(self): + if self._get_data_length(): + if len(self.df) == 1: + L = len(self.df["index"].values[0].strip()) + if L: + return { + "msg": f"Only one sample. Index length is {L}", + "status": "Success", + } + else: + return { + "msg": f"Only one sample. Index length is {L}", + "status": "Error", + } + + diff = self.df["index"].apply(lambda x: len(x)).std() + if diff == 0: + return { + "msg": "Indices length in I7 have same lengths", + "status": "Success", + } + else: + return { + "msg": "Indices length in I7 have different lengths", + "status": "Error", + } + else: + return {"msg": "Indices length could not be read.", "status": "Warning"} + + def _check_homogene_I5_length(self): + if self._get_data_length(): + + if len(self.df) == 1: + L = len(self.df["index2"].values[0].strip()) + if L: + return { + "msg": f"Only one sample. Index length for I5 is {L}", + "status": "Success", + } + else: + return { + "msg": f"Only one sample. Index length for I5 is {L}", + "status": "Error", + } + diff = self.df["index2"].apply(lambda x: len(x)).std() + if diff == 0: + return { + "msg": "Indices length in I5 have same lengths", + "status": "Success", + } + else: + return { + "msg": "Indices length in I5 have different lengths", + "status": "Error", + } + else: + return {"msg": "Indices length could not be read. ", "status": "Warning"} + + def _check_homogene_I5_and_I7_length(self): + # this is a warning only since you may have custom index + lengths = [len(x) for x in self.df["index"]] + [ + len(x) for x in self.df["index2"] + ] + lengths = list(set(lengths)) + L = lengths[0] + + if len(lengths) == 1: + return { + "msg": f"I5 and I7 have coherent length of {L}", + "status": "Success", + } + else: + return {"msg": "I5 and I7 have different lengths", "status": "Warning"} + + return {"msg": "Indices length could not be read. ", "status": "Success"} + + def _check_data_section_csv_format(self): + + N = len(set([x.count(",") for x in self.sections["Data"]])) + + if N == 1: # looks correct + if len(self.sections["Data"]) == 1: + return { + "name": "check_data_section_csv_format", + "msg": "The [Data] section CSV format looks empty. Remove if of fill it.", + "status": "Error", + } + else: + return { + "name": "check_data_section_csv_format", + "msg": "The [Data] section CSV format looks correct.", + "status": "Success", + } + elif N == 0: + return { + "name": "check_data_section_csv_format", + "msg": "The [Data] section CSV format looks empty. Remove it of fill it", + "status": "Error", + } + else: + lengths = set([x.count(",") for x in self.sections["Data"]]) + return { + "name": "check_data_section_csv_format", + "msg": f"The [Data] section has lines with different number of entries {lengths}. Probably missing or commas in the [Data] section.", + "status": "Error", + } + + def _check_semi_column_presence(self): + with open(self.filename, "r") as fp: + line_count = 1 + for line in fp.readlines(): + if line.rstrip().endswith(";"): + return { + "name": "check_semi_column_presence", + "msg": f"suspicous ; at the end of line ({line_count})", + "status": "Error", + } + line_count += 1 + return { + "name": "check_semi_column_presence", + "msg": "No extra semi column found.", + "status": "Success", + } + + def _check_alpha_numerical(self): + for column in ["sample_id", "sample_name", "sample_project"]: + if column not in self.df.columns: + continue + for i, x in enumerate(self.df[column].values): + status = str(x).replace("-", "").replace("_", "").isalnum() + if status is False: + msg = f"type error: wrong {column} name in [Data] section (line {i+1}). Must be made of alpha numerical characters, _, and - characters only. Found {x}" + return { + "msg": msg, + "name": "check_alpha_numerical", + "status": "Error", + } + return { + "name": "check_alpha_numerical", + "msg": "sample names and ID looks correct in the Sample_ID, Sample_Name, and Project column (alpha numerical and - or _ characters)", + "status": "Success", + } + + def validate(self): + """This method checks whether the sample sheet is correctly formatted + + Checks for: + * presence of ; at the end of lines indicated an edition with excel that + wrongly transformed the data into a pure CSV file + * inconsistent numbers of columns in the [DATA] section, which must be + CSV-like section + * Extra lines at the end are ignored + * special characters are forbidden except - and _ + * checks for Sample_ID column uniqueness + * checks for index uniqueness (if single index) + * checks for combo of dual indices uniqueness + * checks that sample names are unique + + and raise a SystemExit error on the first found error. + + """ + # aggregates all checks + checks = self.checker() + + # Stop after first error + for check in checks: + if check["status"] == "Error": + sys.exit("\u274c " + str(check["msg"])) + + def _get_settings(self): + data = {} + for line in self.sections["Settings"]: + key, value = line.split(",") + data[key] = value + return data + + settings = property(_get_settings) + + def _get_header(self): + data = {} + for line in self.sections["Header"]: + key, value = line.split(",", 1) + data[key] = value + return data + + header = property(_get_header) + + def _get_instrument(self): + try: + return self.header["Instrument Type"] + except KeyError: + return None + + instrument = property(_get_instrument, doc="returns instrument name") + + def _get_adapter_kit(self): + try: + return self.header["Index Adapters"] + except KeyError: + return None + + index_adapters = property(_get_adapter_kit, doc="returns index adapters") + + def to_fasta(self, adapter_name=""): + """Extract adapters from [Adapter] section and print them as a fasta file""" + ar1 = self.settings["Adapter"] + try: + ar2 = self.settings["AdapterRead2"] + except KeyError: + ar2 = "" + + for name, index in zip(self.df["i7_index_id"], self.df["index"]): + read = f"{ar1}{index}{ar2}" + frmt = {"adapter": adapter_name, "name": name, "index": index} + print(">{adapter}_index_{name}|name:{name}|seq:{index}".format(**frmt)) + print(read) + + if "index2" in self.df.columns: + for name, index in zip(self.df["i5_index_id"], self.df["index2"]): + read = f"{ar1}{index}{ar2}" + frmt = {"adapter": adapter_name, "name": name, "index": index} + print(">{adapter}_index_{name}|name:{name}|seq:{index}".format(**frmt)) + print(read) + + def quick_fix(self, output_filename): + """Fix sample sheet + + Tyical error is when users save the samplesheet as CSV file in excel. + This may add trailing ; characters at the end of section, which raises error + in bcl2fastq. + + """ + found_data = False + with open(self.filename) as fin: + with open(output_filename, "w") as fout: + for line in fin.readlines(): + if line.startswith("[Data]"): + found_data = True + + if found_data: + line = line.replace(";", ",") + else: + line = line.strip().rstrip(";") + line = line.replace(";", ",") + line = line.strip().rstrip(",") + fout.write(line.strip("\n") + "\n") + + +class BCLConvert: + """BCLconvert is a replacement for the bcl2fastq software + + The format are quite similar but there are differences as explained in the + official Illumina document. + + https://support.illumina.com/sequencing/sequencing_software/bcl-convert/compatibility.html + https://knowledge.illumina.com/software/general/software-general-reference_material-list/000003710 + + Some differences with bcl2fastq + + - sample ID compulsary --> error if absent + - sample Name ignored --> warning if present + - fastq header. filter is set to N and control bit to 0. missing instrument is not supported -->error + - V1 and V2 accepted. + - Note: At least one Sample_ID is required in the Data section of the Sample Sheet. --> error + + for barcode-mismatch, this is now in the samplesheet with + + BarcodeMismatchIndex1, # + BarcodeMismatchIndex2, # + + for trmimming, masking: + + Adapter --> Changed to AdapterRead1 + AdapterRead2 unchanged + + Many options that were on command lines are now in the sample sheet + e.g. --minimum-trimmed-read-length is now MinimumTrimmedReadLength + + Supports both [Settings] and [settings]. Neither are required. + + Supports only [BCLConvert_Settings]. Required. + + """ + + def __init__(self): + pass + + # obsolet names are + # Adapter,TrimAdapter,MaskAdapter,MaskAdapterRead2,TrimAdapter,Read1StartFromCycle,Read1EndWithCycle, + # Read1UMIStartFromCycle,Read1UMILength,Read1StartFromCycle,,Read2UMIStartFromCycle,Read2UMILength,Read2StartFromCycle + + +class IEM(SampleSheet): + def __init__(self, filename): + super().__init__(filename) + logger.warning("IEM class is deprecated. Use SampleShee instead.") diff --git a/samplesheet_validator.def b/samplesheet_validator.def new file mode 100644 index 0000000..53f9f70 --- /dev/null +++ b/samplesheet_validator.def @@ -0,0 +1,12 @@ +Bootstrap: docker +From: python:3.14-slim + +%files + app /app + +%post + python3 -m pip install -U uv + uv pip install --system -r /app/requirements.txt + +%startscript + cd /app && python3 -m streamlit run --server.headless true --server.address {{ SERVER_ADDRESS }} app.py