4627485829

Build Type

push

github-actions

Committed by karlnyr

Commit Message

remova conda stuff, expand aliases

Run Details

501 of 941 relevant lines covered (53.24%)

0.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/demux/cli/basemask.py

""" create basemask for demultiplexing based on run parameters """

import logging
import xml.etree.cElementTree as xml_etree
from pathlib import Path

import click

from ..utils import HiSeqXSamplesheet, MiseqSamplesheet, NIPTSamplesheet, Samplesheet

EMPTY_STRING = ""
LOG = logging.getLogger(__name__)


@click.group()
def basemask():
    """Samplesheet commands"""


@basemask.command()
@click.argument("rundir")
@click.option("-l", "--lane", help="lane number")
@click.option(
    "-a",
    "--application",
    type=click.Choice(["miseq", "nipt", "wes", "wgs"]),
    help="sequencing type",
)
def create(rundir, lane, application):
    """Create a basemask based on SampleSheet.csv and runParameters"""

    # runParameters.xml
    def parse_run_parameters():
        """parse the run parameters file"""
        return xml_etree.parse(Path(rundir).joinpath("runParameters.xml"))

    def create_basemask(sheet):
        """create the bcl2fastq basemask"""
        run_params_tree = parse_run_parameters()
        read1 = int(run_params_tree.findtext("Setup/Read1"))
        read2 = int(run_params_tree.findtext("Setup/Read2"))
        indexread1 = int(run_params_tree.findtext("Setup/IndexRead1"))
        indexread2 = int(run_params_tree.findtext("Setup/IndexRead2"))

        lines = [line for line in sheet.lines_per_column("lane", lane)]

        # get the index lengths
        index1 = lines[0]["index"]
        index2 = lines[0]["index2"] if "index2" in lines[0] else EMPTY_STRING

        # index1 basemask
        index1_n = "n" * (indexread1 - len(index1))
        basemask_index1 = "I" + str(len(index1)) + index1_n

        # index2 basemask
        if indexread2 == 0:
            click.echo(f"Y{read1},{basemask_index1},Y{read2}")
        else:
            index2_n = "n" * (indexread2 - len(index2))
            if len(index2) > 0:
                basemask_index2 = "I" + str(len(index2)) + index2_n
            else:
                basemask_index2 = index2_n
            click.echo(f"Y{read1},{basemask_index1},{basemask_index2},Y{read2}")

    def get_application_sheet(application):
        """parse the samplesheet in the runs directory, based on the type of application"""
        samplesheet = Path(rundir).joinpath("SampleSheet.csv")
        sheet_map = {
            "nipt": NIPTSamplesheet,
            "wes": Samplesheet,
            "miseq": MiseqSamplesheet,
            "wgs": HiSeqXSamplesheet,
        }
        sheet = sheet_map[application](samplesheet)

        return sheet

    def create_application_basemask(application):
        """determine the basemask"""
        create_basemask(get_application_sheet(application))

    create_application_basemask(application)

1	""" create basemask for demultiplexing based on run parameters """
2
3	import logging	×
4	import xml.etree.cElementTree as xml_etree	×
5	from pathlib import Path	×
6
7	import click	×
8
9	from ..utils import HiSeqXSamplesheet, MiseqSamplesheet, NIPTSamplesheet, Samplesheet	×
10
11	EMPTY_STRING = ""	×
12	LOG = logging.getLogger(__name__)	×
13
14
15	@click.group()	×
16	def basemask():
17	"""Samplesheet commands"""
18
19
20	@basemask.command()	×
21	@click.argument("rundir")	×
22	@click.option("-l", "--lane", help="lane number")	×
23	@click.option(	×
24	"-a",
25	"--application",
26	type=click.Choice(["miseq", "nipt", "wes", "wgs"]),
27	help="sequencing type",
28	)
29	def create(rundir, lane, application):
30	"""Create a basemask based on SampleSheet.csv and runParameters"""
31
32	# runParameters.xml
33	def parse_run_parameters():	×
34	"""parse the run parameters file"""
35	return xml_etree.parse(Path(rundir).joinpath("runParameters.xml"))	×
36
37	def create_basemask(sheet):	×
38	"""create the bcl2fastq basemask"""
39	run_params_tree = parse_run_parameters()	×
40	read1 = int(run_params_tree.findtext("Setup/Read1"))	×
41	read2 = int(run_params_tree.findtext("Setup/Read2"))	×
42	indexread1 = int(run_params_tree.findtext("Setup/IndexRead1"))	×
43	indexread2 = int(run_params_tree.findtext("Setup/IndexRead2"))	×
44
45	lines = [line for line in sheet.lines_per_column("lane", lane)]	×
46
47	# get the index lengths
48	index1 = lines[0]["index"]	×
49	index2 = lines[0]["index2"] if "index2" in lines[0] else EMPTY_STRING	×
50
51	# index1 basemask
52	index1_n = "n" * (indexread1 - len(index1))	×
53	basemask_index1 = "I" + str(len(index1)) + index1_n	×
54
55	# index2 basemask
56	if indexread2 == 0:	×
57	click.echo(f"Y{read1},{basemask_index1},Y{read2}")	×
58	else:
59	index2_n = "n" * (indexread2 - len(index2))	×
60	if len(index2) > 0:	×
61	basemask_index2 = "I" + str(len(index2)) + index2_n	×
62	else:
63	basemask_index2 = index2_n	×
64	click.echo(f"Y{read1},{basemask_index1},{basemask_index2},Y{read2}")	×
65
66	def get_application_sheet(application):	×
67	"""parse the samplesheet in the runs directory, based on the type of application"""
68	samplesheet = Path(rundir).joinpath("SampleSheet.csv")	×
69	sheet_map = {	×
70	"nipt": NIPTSamplesheet,
71	"wes": Samplesheet,
72	"miseq": MiseqSamplesheet,
73	"wgs": HiSeqXSamplesheet,
74	}
75	sheet = sheet_map[application](samplesheet)	×
76
77	return sheet	×
78
79	def create_application_basemask(application):	×
80	"""determine the basemask"""
81	create_basemask(get_application_sheet(application))	×
82
83	create_application_basemask(application)	×

Clinical-Genomics / demultiplexing / 4627485829

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous