fastq2bcl convert fastq files in a bcl2fastq-able run directory.
# This file is execfile()d with the current directory set to its containing dir.
#
# This file only contains a selection of the most common options. For a full
# list see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
#
# All configuration values have a default; values that are commented out
# serve to show the default.

import os
import sys
import shutil

# -- Path setup --------------------------------------------------------------

__location__ = os.path.dirname(__file__)

# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
sys.path.insert(0, os.path.join(__location__, "../src"))

# -- Run sphinx-apidoc -------------------------------------------------------
# This hack is necessary since RTD does not issue `sphinx-apidoc` before running
# `sphinx-build -b html . _build/html`. See Issue:
# https://github.com/readthedocs/readthedocs.org/issues/1139
# DON'T FORGET: Check the box "Install your project inside a virtualenv using
# setup.py install" in the RTD Advanced Settings.
# Additionally it helps us to avoid running apidoc manually

try:  # for Sphinx >= 1.7
    from sphinx.ext import apidoc
except ImportError:
    from sphinx import apidoc

output_dir = os.path.join(__location__, "api")
module_dir = os.path.join(__location__, "../src/fastq2bcl")
try:
    shutil.rmtree(output_dir)
except FileNotFoundError:
    pass

try:
    import sphinx

    cmd_line = f"sphinx-apidoc --implicit-namespaces -f -o {output_dir} {module_dir}"

    args = cmd_line.split(" ")
    if tuple(sphinx.__version__.split(".")) >= ("1", "7"):
        # This is a rudimentary parse_version to avoid external dependencies
        args = args[1:]

    apidoc.main(args)
except Exception as e:
    print("Running `sphinx-apidoc` failed!\n{}".format(e))

# -- General configuration ---------------------------------------------------

# If your documentation needs a minimal Sphinx version, state it here.
# needs_sphinx = '1.0'

# Add any Sphinx extension module names here, as strings. They can be extensions
# coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
extensions = [
    "sphinx.ext.autodoc",
    "sphinx.ext.intersphinx",
    "sphinx.ext.todo",
    "sphinx.ext.autosummary",
    "sphinx.ext.viewcode",
    "sphinx.ext.coverage",
    "sphinx.ext.doctest",
    "sphinx.ext.ifconfig",
    "sphinx.ext.mathjax",
    "sphinx.ext.napoleon",
]

# Add any paths that contain templates here, relative to this directory.
templates_path = ["_templates"]

# The suffix of source filenames.
source_suffix = ".rst"

# The encoding of source files.
# source_encoding = 'utf-8-sig'

# The master toctree document.
master_doc = "index"

# General information about the project.
project = "fastq2bcl"
copyright = "2023, Davide Rambaldi"

# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
# built documents.
#
# version: The short X.Y version.
# release: The full version, including alpha/beta/rc tags.
# If you don’t need the separation provided between version and release,
# just set them both to the same value.
try:
    from fastq2bcl import __version__ as version
except ImportError:
    version = ""

if not version or version.lower() == "unknown":
    version = os.getenv("READTHEDOCS_VERSION", "unknown")  # automatically set by RTD

release = version

# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
# language = None

# There are two options for replacing |today|: either, you set today to some
# non-false value, then it is used:
# today = ''
# Else, today_fmt is used as the format for a strftime call.
# today_fmt = '%B %d, %Y'

# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
exclude_patterns = ["_build", "Thumbs.db", ".DS_Store", ".venv"]

# The reST default role (used for this markup: `text`) to use for all documents.
# default_role = None

# If true, '()' will be appended to :func: etc. cross-reference text.
# add_function_parentheses = True

# If true, the current module name will be prepended to all description
# unit titles (such as .. function::).
# add_module_names = True

# If true, sectionauthor and moduleauthor directives will be shown in the
# output. They are ignored by default.
# show_authors = False

# The name of the Pygments (syntax highlighting) style to use.
pygments_style = "sphinx"

# A list of ignored prefixes for module index sorting.
# modindex_common_prefix = []

# If true, keep warnings as "system message" paragraphs in the built documents.
# keep_warnings = False

# If this is True, todo emits a warning for each TODO entries. The default is False.
todo_emit_warnings = True


# -- Options for HTML output -------------------------------------------------

# The theme to use for HTML and HTML Help pages.  See the documentation for
# a list of builtin themes.
html_theme = "alabaster"

# Theme options are theme-specific and customize the look and feel of a theme
# further.  For a list of options available for each theme, see the
# documentation.
html_theme_options = {"sidebar_width": "300px", "page_width": "1200px"}

# Add any paths that contain custom themes here, relative to this directory.
# html_theme_path = []

# The name for this set of Sphinx documents.  If None, it defaults to
# "<project> v<release> documentation".
# html_title = None

# A shorter title for the navigation bar.  Default is the same as html_title.
# html_short_title = None

# The name of an image file (relative to this directory) to place at the top
# of the sidebar.
# html_logo = ""

# The name of an image file (within the static path) to use as favicon of the
# docs.  This file should be a Windows icon file (.ico) being 16x16 or 32x32
# pixels large.
# html_favicon = None

# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ["_static"]

# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
# html_last_updated_fmt = '%b %d, %Y'

# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
# html_use_smartypants = True

# Custom sidebar templates, maps document names to template names.
# html_sidebars = {}

# Additional templates that should be rendered to pages, maps page names to
# template names.
# html_additional_pages = {}

# If false, no module index is generated.
# html_domain_indices = True

# If false, no index is generated.
# html_use_index = True

# If true, the index is split into individual pages for each letter.
# html_split_index = False

# If true, links to the reST sources are added to the pages.
# html_show_sourcelink = True

# If true, "Created using Sphinx" is shown in the HTML footer. Default is True.
# html_show_sphinx = True

# If true, "(C) Copyright ..." is shown in the HTML footer. Default is True.
# html_show_copyright = True

# If true, an OpenSearch description file will be output, and all pages will
# contain a <link> tag referring to it.  The value of this option must be the
# base URL from which the finished HTML is served.
# html_use_opensearch = ''

# This is the file name suffix for HTML files (e.g. ".xhtml").
# html_file_suffix = None

# Output file base name for HTML help builder.
htmlhelp_basename = "fastq2bcl-doc"


# -- Options for LaTeX output ------------------------------------------------

latex_elements = {
    # The paper size ("letterpaper" or "a4paper").
    # "papersize": "letterpaper",
    # The font size ("10pt", "11pt" or "12pt").
    # "pointsize": "10pt",
    # Additional stuff for the LaTeX preamble.
    # "preamble": "",
}

# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title, author, documentclass [howto/manual]).
latex_documents = [
    ("index", "user_guide.tex", "fastq2bcl Documentation", "Davide Rambaldi", "manual")
]

# The name of an image file (relative to this directory) to place at the top of
# the title page.
# latex_logo = ""

# For "manual" documents, if this is true, then toplevel headings are parts,
# not chapters.
# latex_use_parts = False

# If true, show page references after internal links.
# latex_show_pagerefs = False

# If true, show URL addresses after external links.
# latex_show_urls = False

# Documents to append as an appendix to all manuals.
# latex_appendices = []

# If false, no module index is generated.
# latex_domain_indices = True

# -- External mapping --------------------------------------------------------
python_version = ".".join(map(str, sys.version_info[0:2]))
intersphinx_mapping = {
    "sphinx": ("https://www.sphinx-doc.org/en/master", None),
    "python": ("https://docs.python.org/" + python_version, None),
    "matplotlib": ("https://matplotlib.org", None),
    "numpy": ("https://numpy.org/doc/stable", None),
    "sklearn": ("https://scikit-learn.org/stable", None),
    "pandas": ("https://pandas.pydata.org/pandas-docs/stable", None),
    "scipy": ("https://docs.scipy.org/doc/scipy/reference", None),
    "setuptools": ("https://setuptools.pypa.io/en/stable/", None),
    "pyscaffold": ("https://pyscaffold.org/en/stable", None),
}

print(f"loading configurations for {project} {version} ...", file=sys.stderr)

docs/conf.py
tucano/fastq2bcl
"""
    Setup file for fastq2bcl.
    Use setup.cfg to configure your project.

    This file was generated with PyScaffold 4.5.
    PyScaffold helps you to put up the scaffold of your new Python project.
    Learn more under: https://pyscaffold.org/
"""
from setuptools import setup

if __name__ == "__main__":
    try:
        setup(use_scm_version={"version_scheme": "no-guess-dev"})
    except:  # noqa
        print(
            "\n\nAn error occurred while building the project, "
            "please ensure you have the most updated version of setuptools, "
            "setuptools_scm and wheel with:\n"
            "   pip install -U setuptools setuptools_scm wheel\n\n"
        )
        raise

setup.py
tucano/fastq2bcl
"""
CLI for fastq2bcl app

``[options.entry_points]`` section in ``setup.cfg``::

    console_scripts =
        fastq2bcl = fastq2bcl.cli:run

Then run ``pip install .`` (or ``pip install -e .`` for editable mode)
which will install the command ``fastq2bcl`` inside your current environment.

References:
    - https://setuptools.pypa.io/en/latest/userguide/entry_point.html
    - https://pip.pypa.io/en/stable/reference/pip_install
"""

import argparse
import logging
import sys
import os
import re
import textwrap
from pathlib import Path
from rich import print, pretty

from fastq2bcl import __version__
from fastq2bcl.parser import parse_seqdesc_fields
from fastq2bcl.reader import read_first_record, read_fastq_files, get_mask_from_files
from fastq2bcl.writer import (
    write_run_info_xml,
    write_filter,
    write_control,
    write_locs,
    write_bcls_and_stats,
)

__author__ = "Davide Rambaldi"
__copyright__ = "Davide Rambaldi"
__license__ = "MIT"

_logger = logging.getLogger(__name__)

# ---- Python API ----
# The functions defined in this section can be imported by users in their
# Python scripts/interactive interpreter, e.g. via
# `from fastq2bcl.skeleton import fib`,
# when using this Python module as a library.


def fastq2bcl(
    outdir,
    r1,
    r2=None,
    i1=None,
    i2=None,
    mask_string=None,
    exclude_umi=False,
    exclude_index=False,
):
    """fastq2bcl function call

    :param outdir: output directory to create run flowcell fake dir
    :param r1: R1 fastq.gz
    :param r2: R2 fastq.gz
    :param i1: I1 fastq.gz
    :param i2: I2 fastq.gz

    Content of returned tuple:

    rundir: final absolute path od created rundir
    run_id: generated mock run_id
    seq_fields: fields parsed and validated from first R1 record
    mask_string: mask used to generate RunInfo.xml

    :rtype: tuple
    """

    # First validate outdir
    outdir = Path(outdir).absolute()
    assert outdir.is_dir()
    assert os.access(outdir, os.W_OK)

    _logger.info(f"Output directory: {outdir}")

    # Validate R1 and extract first read
    r1 = Path(r1)
    assert r1.is_file()
    first_record = read_first_record(r1)
    seqdesc_fields = parse_seqdesc_fields(first_record.description)
    _logger.info(f"first record seq length: {len(first_record.seq)}")
    _logger.info(f"first record sequence: {str(first_record.seq)}")
    _logger.info(f"first record seqdesc fields: {seqdesc_fields}")

    print(f"[green]First sequence length[/green]: {len(first_record.seq)}")
    print(f"[green]First sequence[/green]:")
    print(textwrap.fill(str(first_record.seq), 50))

    # SEE LSO docs/flow.drawio.png
    # CHECK INDEX
    if seqdesc_fields["index"] != "1":
        if exclude_index:
            print(
                f"[red]Founded INDEX sequence in first record[/red] {seqdesc_fields['index']}",
                f"[red]index sequences will NOT be included in the cycles[/red]",
            )
        else:
            print(
                f"[green]Founded INDEX sequence in first record[/green]: {seqdesc_fields['index']}",
                f"[green]index sequences will be included in the cycles[/green]",
            )

    # CHECK UMI
    if seqdesc_fields["UMI"] != None:
        if exclude_umi:
            print(
                f"[red]Founded UMI sequence in first record[/red] {seqdesc_fields['UMI']}",
                f"[red]umi sequences will NOT be included in the cycles[/red]",
            )
        else:
            print(
                f"[green]Founded UMI sequence in first record[/green]: {seqdesc_fields['UMI']}",
                f"[green]umi sequences will be included in the cycles[/green]",
            )

    # RUNDIR
    run_id = mock_run_id(seqdesc_fields)
    rundir = Path.joinpath(outdir, run_id)

    print(f"[green]RUNDIR[/green]: {rundir}")

    if not mask_string:
        # get cycles string from files
        mask_string = get_mask_from_files(r1, r2, i1, i2, exclude_umi, exclude_index)
        _logger.info(f"mask string from files: {mask_string}")

    print(f"[green]MASK[/green]: {mask_string}")

    # READ DATA {"sequences": sequences, "positions": positions}
    # TODO auto write indexes from UMI and index
    sequences, positions = read_fastq_files(r1, r2, i1, i2, exclude_umi, exclude_index)

    # SET MASK FROM STRING
    mask = set_mask(mask_string)

    # WRITE RUN INFO
    _logger.info(f"Writing RunInfo.mxl to dir: {rundir}")
    run_info = write_run_info_xml(
        rundir,
        run_id,
        seqdesc_fields["run_number"],
        seqdesc_fields["flowcell_id"],
        seqdesc_fields["instrument"],
        mask,
    )

    print(f"[green]RunInfo.xml:[/green]:\n", run_info)

    # WRITE FILTER
    print(f"[bold magenta]Writing filter file [/bold magenta]")
    _logger.info(
        f"Writing filter file to dir: {rundir} with cluster count: {len(sequences)}"
    )
    write_filter(rundir, len(sequences))

    # WRITE CONTROL
    print(f"[bold magenta]Writing control file [/bold magenta]")
    _logger.info(
        f"Writing control file to dir: {rundir} with cluster count: {len(sequences)}"
    )
    write_control(rundir, len(sequences))

    # WRITE LOCATIONS
    print(f"[bold magenta]Writing location file [/bold magenta]")
    _logger.info(f"Writing {len(positions)} locations to dir: {rundir}")
    write_locs(rundir, positions)

    # WRITE BCL AND STATS
    print(f"[bold magenta]Writing cycles files [/bold magenta]")
    _logger.info(f"Writing {len(sequences)} sequences bcl and stats to dir: {rundir}")

    write_bcls_and_stats(rundir, sequences)

    return (run_id, rundir, seqdesc_fields, mask_string)


def mock_run_id(fields):
    """
    Mock the run directory id and Path
    """
    run_id = (
        "YYMMDD_"
        + fields["instrument"]
        + "_"
        + fields["run_number"].zfill(4)
        + "_"
        + fields["flowcell_id"]
    )
    return run_id


def set_mask(mask_string):
    if mask_string:
        mask = []
        regexp_mask = r"([0-9]+[NY])([0-9]+[NY])?([0-9]+[NY])?([0-9]+[NY])?"
        m = re.match(regexp_mask, mask_string)
        if not m:
            raise ValueError(f"Incorrect mask parse: {mask_string}")
        reads = [g for g in m.groups() if g != None]
        for g_idx in range(len(reads)):
            read = re.match(r"([0-9]+)([YN])", reads[g_idx])
            mask.append(
                {
                    "cycles": read.groups()[0],
                    "index": read.groups()[1],
                    "id": str(g_idx + 1),
                }
            )
        return mask
    else:
        raise ValueError(f"Incorrect mask string: {mask_string}")


# ---- CLI ----
# The functions defined in this section are wrappers around the main Python
# API allowing them to be called directly from the terminal as a CLI
# executable/script.


def parse_args(args):
    """Parse command line parameters

    Args:
      args (List[str]): command line parameters as list of strings
          (for example  ``["--help"]``).

    Returns:
      :obj:`argparse.Namespace`: command line parameters namespace
    """
    parser = argparse.ArgumentParser(
        description="Convert fastq.gz reads and metadata in a bcl2fastq-able run directory"
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"fastq2bcl {__version__}",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        dest="loglevel",
        help="set loglevel to INFO",
        action="store_const",
        const=logging.INFO,
    )
    parser.add_argument(
        "-vv",
        "--very-verbose",
        dest="loglevel",
        help="set loglevel to DEBUG",
        action="store_const",
        const=logging.DEBUG,
    )
    parser.add_argument(
        "-m", "--mask", dest="mask", help="define mask in format 110N10Y10Y110N"
    )

    parser.add_argument(
        "-r1",
        "--read-1",
        dest="r1",
        help="fastq.gz with R1 reads",
        metavar="R1",
        required=True,
    )

    parser.add_argument(
        "-r2",
        "--read-2",
        dest="r2",
        help="fastq.gz with R2 reads (optional)",
        metavar="R2",
    )

    parser.add_argument(
        "-i1",
        "--index-1",
        dest="i1",
        help="fastq.gz with I1 reads (optional)",
        metavar="I1",
    )

    parser.add_argument(
        "-i2",
        "--index-2",
        dest="i2",
        help="fastq.gz with I2 reads (optional)",
        metavar="I2",
    )

    parser.add_argument(
        "-o",
        "--outdir",
        dest="outdir",
        help="Set the output directory for mocked run. default: cwd",
        default=os.getcwd(),
    )
    parser.add_argument(
        "--exclude-umi",
        dest="exclude_umi",
        help="Do not write UMI from the R1 and R2 fastq reads to the cycles",
        action="store_true",
    )
    parser.add_argument(
        "--exclude-index",
        dest="exclude_index",
        help="Do not write Index from the R1 and R2 fastq reads to the cycles",
        action="store_true",
    )
    return parser.parse_args(args)


def setup_logging(loglevel):
    """Setup basic logging

    Args:
      loglevel (int): minimum loglevel for emitting messages
    """
    logformat = "[%(asctime)s] %(levelname)s:%(name)s:%(message)s"
    logging.basicConfig(
        level=loglevel, stream=sys.stdout, format=logformat, datefmt="%Y-%m-%d %H:%M:%S"
    )


def main(args):
    """Wrapper allowing :func:`fastq2bcl` to be called with string arguments in a CLI fashion

    Instead of returning the value from :func:`fastq2bcl`, it prints the result to the
    ``stdout`` in a nicely formatted message.

    Args:
      args (List[str]): command line parameters as list of strings
          (for example  ``["--verbose", "42"]``).
    """
    pretty.install()

    args = parse_args(args)
    setup_logging(args.loglevel)
    _logger.info("Starting application...")
    _logger.info(f"User defined mask: {args.mask}")
    _logger.info(f"Input files: R1={args.r1} R2={args.r2} I1={args.i1} I2={args.i2}")

    print("[bold green]fastq2bcl[/bold green]")
    print("Args:", args)

    # call fastq2bcl
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        args.outdir,
        args.r1,
        args.r2,
        args.i1,
        args.i2,
        args.mask,
        args.exclude_umi,
        args.exclude_index,
    )

    _logger.info("Script ends here")


def run():
    """Calls :func:`main` passing the CLI arguments extracted from :obj:`sys.argv`

    This function can be used as entry point to create console scripts with setuptools.
    """
    main(sys.argv[1:])


if __name__ == "__main__":
    # ^  This is a guard statement that will prevent the following code from
    #    being executed in the case someone imports this file instead of
    #    executing it as a script.
    #    https://docs.python.org/3/library/__main__.html

    # After installing your project with pip, users can also run your Python
    # modules as scripts via the ``-m`` flag, as defined in PEP 338::
    #
    #     python -m fastq2bcl.skeleton 42
    #
    run()

src/fastq2bcl/cli.py
tucano/fastq2bcl
import re
import logging

_logger = logging.getLogger(__name__)


def parse_seqdesc_fields(txt):
    """
    Parse the SeqIO description field using named groups.
    """
    regxp = re.compile(
        r"(?P<instrument>[A-Za-z0-9_]+):"
        + r"(?P<run_number>[0-9]+):"
        + r"(?P<flowcell_id>[A-Za-z0-9-]+):"
        + r"(?P<lane>[0-9]+):"
        + r"(?P<tile>[0-9]+):?"
        + r"(?P<x_pos>[0-9]+)?:?"
        + r"(?P<y_pos>[0-9]+)?:?"
        + r"(?P<UMI>[A-Z-]+)?"
        + r"\s"
        + r"(?P<read>[0-9]+):"
        + r"(?P<is_filtered>[YN]+):"
        + r"(?P<control_number>[0-9]+):"
        + r"(?P<index>[0-9A-Z+]+)"
    )

    match = re.match(regxp, txt)
    if not match:
        raise ValueError(f"Sequence identifier not recognized: {txt}")

    return validate_fields(match.groupdict())


def validate_fields(fields):
    """
    Validate the fields extracted from SeqIO description
    """
    valid_keys = [
        "instrument",
        "run_number",
        "flowcell_id",
        "lane",
        "tile",
        "x_pos",
        "y_pos",
        "UMI",
        "read",
        "is_filtered",
        "control_number",
        "index",
    ]
    _logger.info(f"Verifying keys ...")

    for key in valid_keys:
        if not fields[key]:
            if key == "UMI" and fields[key] == None:
                _logger.info(f"Found None value for optional key {key}. This is ok.")
            else:
                raise ValueError(f"Requested Key {key} not Found in fastq description")

    return fields

src/fastq2bcl/parser.py
tucano/fastq2bcl
import logging
import gzip
from fastq2bcl.parser import parse_seqdesc_fields
from Bio import SeqIO
from rich import print
import sys

_logger = logging.getLogger(__name__)


def read_first_record(fastq_file):
    """
    Validate fastq.gz r1 file and extract first read
    """
    _logger.info(f"Opening gz file {fastq_file}")
    with gzip.open(fastq_file, "rt") as fastq_fh:
        return next(SeqIO.parse(fastq_fh, "fastq"))


def get_file_handlers(r1, r2, i1, i2):
    """
    Return list of FH
    """
    files_fh = [gzip.open(r1, "rt")]
    if not i1 == None:
        files_fh.append(gzip.open(i1, "rt"))
    if not i2 == None:
        files_fh.append(gzip.open(i2, "rt"))
    if not r2 == None:
        files_fh.append(gzip.open(r2, "rt"))

    return files_fh


def get_mask_from_files(r1, r2, i1, i2, exclude_umi, exclude_index):
    """
    Build a mask string using seq length. In case of index and/or UMI in R1 sequence description, write this length to the Index mask
    """
    record_1 = read_first_record(r1)
    seq_fields = parse_seqdesc_fields(record_1.description)
    index_1_bases = 0

    # Write R1 mask
    mask = f"{len(record_1.seq)}N"

    print(f"I AM HERE with index {seq_fields['index']} and exclude is {exclude_index}")

    # check errors on index for R1
    if seq_fields["index"] != "1" and not exclude_index:
        if i1 != None or i2 != None:
            raise ValueError(
                "Usage of index from sequence desc and I1 and I2 files at the same time is not supported"
            )
        # continue and write to index I1 length TODO I2 for double index
        print(f"LENGTH INDEX {seq_fields['index']}")
        index_1_bases += len(seq_fields["index"])

    # check errors on UMI for R1
    if seq_fields["UMI"] != None and not exclude_umi:
        if i1 != None or i2 != None:
            raise ValueError(
                "Usage of UMI from sequence desc and I1 and I2 files at the same time is not supported"
            )
        # continue and write to index I1
        index_1_bases += len(seq_fields["UMI"])

    # Write index I1 based on UMI and index
    if index_1_bases > 0:
        mask += f"{index_1_bases}Y"

    # Write indexes
    if i1 != None:
        index_1 = read_first_record(i1)
        mask += f"{len(index_1.seq)}Y"

    if i2 != None:
        index_2 = read_first_record(i2)
        mask += f"{len(index_2.seq)}Y"

    # Write R2 record
    if r2 != None:
        record_2 = read_first_record(r2)
        # finally add R2 to mask
        mask += f"{len(record_2.seq)}N"

    return mask


def read_fastq_files(r1, r2, i1, i2, exclude_umi, exclude_index):
    """
    Read fastq files R1-R2 with I1 and I2 and return only the data we need
    """
    # return a list of tuple with seq, qual
    # and a list of tuple for pos with x and y
    # SINGLE R1
    # sequences = [('AAAA',1111)]
    # positions = [(1,1)]
    #
    # in case of multiple files R1-R2:
    # PAIR R1-R2
    # sequences = [('AAAABBBB',11111111)]
    # positions = [(1,1)]
    #
    # I need way to handle multiple files and merge them in a single with exitstack
    # Ref https://docs.python.org/3/library/contextlib.html#contextlib.ExitStack

    # build a list of iterators
    file_handlers = get_file_handlers(r1, r2, i1, i2)
    seq_iterators = [SeqIO.parse(fh, "fastq") for fh in file_handlers]

    # output Lists
    sequences = []
    positions = []

    try:
        # iterate over the R1 iterator
        for r1_record in seq_iterators[0]:
            # call next in additional iterators
            opt_data = [next(iterator) for iterator in seq_iterators[1:]]

            # store R1 data
            record_fields = parse_seqdesc_fields(r1_record.description)
            record_id = r1_record.id
            record_seq = str(r1_record.seq)
            record_qual = r1_record.letter_annotations["phred_quality"]

            if not exclude_index and record_fields["index"] != "1":
                _logger.info(f"Reading index field: {record_fields['index']}")
                record_seq += record_fields["index"]
                _logger.info(f"New seq len: {len(record_seq)} seq: {record_seq}")
                record_qual += [40] * len(record_fields["index"])
                _logger.info(f"New qual: {record_qual}")

            # the UMI is quality MAX (40)
            if not exclude_umi and record_fields["UMI"] != None:
                _logger.info(f"Reading umi field: {record_fields['UMI']}")
                record_seq += record_fields["UMI"]
                _logger.info(f"New seq len: {len(record_seq)} seq: {record_seq}")
                record_qual += [40] * len(record_fields["UMI"])
                _logger.info(f"New qual: {record_qual}")

            for opt_record in opt_data:
                if opt_record.id != record_id:
                    raise ValueError(
                        f"Seq ID mismatch for record {opt_record.id} R1 is {record_id}"
                    )
                record_seq += str(opt_record.seq)
                record_qual += opt_record.letter_annotations["phred_quality"]

            # append cluster position
            positions.append((record_fields["x_pos"], record_fields["y_pos"]))
            # append sequence and qual
            sequences.append((record_seq, record_qual))
    finally:
        # close all files
        for file_fh in file_handlers:
            file_fh.close()

    return (sequences, positions)

src/fastq2bcl/reader.py
tucano/fastq2bcl
import logging
import struct
from pathlib import Path

from rich.progress import track

_logger = logging.getLogger(__name__)


def write_run_info_xml(rundir, run_id, run_number, flowcell_id, instrument, mask):
    """
    Write RunInfo.xml
    """

    runinfo = generate_run_info_xml(run_id, run_number, flowcell_id, instrument, mask)
    _logger.info(f"RunInfo.xml:\n{runinfo}")

    # Create directory and write file
    xmlout = Path.joinpath(rundir, "RunInfo.xml")
    xmlout.parent.mkdir(exist_ok=True, parents=True)
    with open(xmlout, "wt") as f_out:
        f_out.write(runinfo)

    return runinfo


def generate_run_info_xml(run_id, run_number, flowcell_id, instrument, mask):
    """
    Generate a valid Runinfo xml file.
    """

    # check mask and write mask
    xml_mask = ""
    for m in mask:
        xml_mask += f"""<Read NumCycles="{m['cycles']}" Number="{m['id']}" IsIndexedRead="{m['index']}" />"""

    xml = f"""<?xml version="1.0"?>
<RunInfo xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" Version="2">
    <Run Id="{run_id}" Number="{run_number}">
        <Flowcell>{flowcell_id}</Flowcell>
        <Instrument>{instrument}</Instrument>
        <Date>YYMMDD</Date>
        <Reads>
            { xml_mask }
        </Reads>
        <FlowcellLayout LaneCount="1" SurfaceCount="1" SwathCount="1" TileCount="1" />
    </Run>
</RunInfo>
"""
    return xml


def write_filter(rundir, cluster_count):
    """
    Write filter
    """
    path = rundir / "Data/Intensities/BaseCalls/L001/s_1_1101.filter"
    path.parent.mkdir(exist_ok=True, parents=True)
    with open(path, "wb") as f_out:
        f_out.write(bytes([0, 0, 0, 0]))
        f_out.write(bytes([3, 0, 0, 0]))
        f_out.write(struct.pack("<I", cluster_count))
        f_out.write(bytes([1] * cluster_count))


def write_control(rundir, cluster_count):
    """
    Write control file
    """
    path = rundir / "Data/Intensities/BaseCalls/L001/s_1_1101.control"
    path.parent.mkdir(exist_ok=True, parents=True)
    with open(path, "wb") as f_out:
        f_out.write(bytes([0, 0, 0, 0]))  # "Zero value (for backwards compatibility)"
        f_out.write(bytes([2, 0, 0, 0]))  # "Format version number"
        f_out.write(struct.pack("<I", cluster_count))  # "Number of clusters"
        f_out.write(bytes([0, 0] * cluster_count))  # two bytes for each cluster


def write_locs(outdir, positions):
    """
    Write locations.

    Args:
        Positions (List(tuple)): is a List of tuple with x and y values
    """
    # From mkdata.sh of bcl2fastq

    # printf '0: 010000000000803f' | xxd -r -g0 > "$locs_filename"
    # printf '0: %.8x' $clusters_count | sed -E 's/0: (..)(..)(..)(..)/0: \4\3\2\1/' | xxd -r -g0 >> "$locs_filename"

    # So with 1 cluster count should be
    # 01 00 00 00    00 00 80 3f
    # 01 00 00 00    CDCC8C3F 9A99993F

    # Source of this is bcl2fastq/src/cxx/lib/data

    # struct Record
    # {
    #     /// \brief X-coordinate.
    #     float x_;
    #     /// \brief y-coordinate.
    #     float y_;
    # }
    path = Path(outdir) / "Data/Intensities/L001/s_1_1101.locs"
    path.parent.mkdir(exist_ok=True, parents=True)
    with open(path, "wb") as f_out:
        f_out.write(bytes([1, 0, 0, 0, 0, 0, 0x80, 0x3F]))
        f_out.write(struct.pack("<I", len(positions)))
        for position in positions:
            f_out.write(encode_loc_bytes(position[0], position[1]))


def encode_loc_bytes(x_pos, y_pos):
    """
    Encode x and y positon.
    FIXME this is not the correct formul according to the bcl2fastq source code.
    """
    x_bytes = struct.pack("<f", (int(x_pos) - 1000) / 10)
    y_bytes = struct.pack("<f", (int(y_pos) - 1000) / 10)
    return x_bytes + y_bytes


def write_bcls_and_stats(outdir, sequences):
    """
    Write bcl cycle files and stats file

    Args:
        outdir: output directory to create run flowcell fake dir
        sequences: list of tuple (basecalls,qualscores)
    """

    # write cluster counts first
    cycles = len(sequences[0][0])
    _logger.info(f"Number of cycles: {cycles} number of clusters: {len(sequences)}")
    for cycle in track(
        range(cycles),
        description="[bold magenta]Initialize bcl with cluster counts ...[/bold magenta]",
    ):
        # print(f"[magenta]cycle {cycle+1}[/magenta]")
        cycledir = outdir / f"Data/Intensities/BaseCalls/L001/C{cycle+1}.1"
        cycledir.mkdir(exist_ok=True, parents=True)
        with open(cycledir / "s_1_1101.bcl", "wb") as f_out:
            f_out.write(struct.pack("<I", len(sequences)))

    # write individual bases across all clusters for each cycle
    for cycle in track(
        range(cycles),
        description="[bold magenta]Writing bcl and stats ...[/bold magenta]",
    ):
        cycledir = outdir / f"Data/Intensities/BaseCalls/L001/C{cycle+1}.1"
        for basecalls, qualscores in sequences:
            bcl_byte = encode_cluster_byte(basecalls[cycle], qualscores[cycle])
            with open(cycledir / "s_1_1101.bcl", "ab") as f_out:
                f_out.write(bcl_byte)
        with open(cycledir / "s_1_1101.stats", "wb") as f_out:
            # can I get away with this?
            f_out.write(bytes([0] * 108))


def encode_cluster_byte(base, qual):
    """
    Encode cluster byte.
    Bits 0-1 are the bases, respectively [A, C, G, T]
    for [0, 1, 2, 3]:
    bits 2-7 are shifted by two bits and contain the quality score.
    All bits 0 in a byte is reserved for no-call.
    """
    if base == "N":
        return bytes([0])  # no call
    qual = qual << 2
    base = ["A", "C", "G", "T"].index(base)
    return bytes([qual | base])

src/fastq2bcl/writer.py
tucano/fastq2bcl
"""
    Dummy conftest.py for fastq2bcl.

    If you don't know what this is for, just leave it empty.
    Read more about conftest.py under:
    - https://docs.pytest.org/en/stable/fixture.html
    - https://docs.pytest.org/en/stable/writing_plugins.html
"""

# import pytest

tests/conftest.py
tucano/fastq2bcl
import pytest
import sys

from fastq2bcl.cli import main, mock_run_id, fastq2bcl, set_mask, run

__author__ = "Davide Rambaldi"
__copyright__ = "Davide Rambaldi"
__license__ = "MIT"

test_fields = {
    "instrument": "M11111",
    "run_number": "222",
    "flowcell_id": "000000000-K9H97",
    "lane": "1",
    "tile": "1101",
    "x_pos": "19304",
    "y_pos": "1328",
    "UMI": "AAACGGG",
    "read": "1",
    "is_filtered": "N",
    "control_number": "0",
    "index": "1",
}

expected_mask_110N = [{"cycles": "110", "index": "N", "id": "1"}]
expected_mask_110N10Y10Y110N = [
    {"cycles": "110", "index": "N", "id": "1"},
    {"cycles": "10", "index": "Y", "id": "2"},
    {"cycles": "10", "index": "Y", "id": "3"},
    {"cycles": "110", "index": "N", "id": "4"},
]


def test_run(capsys):
    """run CLI test"""
    with pytest.raises(SystemExit):
        run()


def test_main_usage(capsys, tmpdir):
    """CLI Tests"""
    # capsys is a pytest fixture that allows asserts against stdout/stderr
    # https://docs.pytest.org/en/stable/capture.html
    main(["-o", str(tmpdir), "-r1", "data/test/01_single/test_single.fastq.gz"])
    captured = capsys.readouterr()
    assert "YYMMDD_M11111_0222_000000000-K9H97" in captured.out


def test_mock_run_id():
    """Mock run id Tests"""
    assert mock_run_id(test_fields) == "YYMMDD_M11111_0222_000000000-K9H97"


def test_fastq2bcl(tmpdir):
    """Fastq2bcl main function Tests"""
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        str(tmpdir), "data/test/01_single/test_single.fastq.gz"
    )
    assert seqdesc_fields["flowcell_id"] == "000000000-K9H97"
    assert run_id == "YYMMDD_M11111_0222_000000000-K9H97"
    assert mask_string == "110N"


def test_fastq2bcl_with_mask(tmpdir):
    """Fastq2bcl main function Tests"""
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        str(tmpdir),
        "data/test/07_pair/R1.fastq.gz",
        "data/test/07_pair/R2.fastq.gz",
        mask_string="309N309N",
    )
    assert seqdesc_fields["flowcell_id"] == "ABCD"
    assert run_id == "YYMMDD_run_0001_ABCD"
    assert mask_string == "309N309N"


def test_set_mask():
    """Test mask generation"""
    assert set_mask("110N") == expected_mask_110N
    assert set_mask("110N10Y10Y110N") == expected_mask_110N10Y10Y110N
    with pytest.raises(ValueError):
        set_mask(None)
    with pytest.raises(ValueError):
        set_mask("100")


def test_fastq2bcl_with_umi(tmpdir):
    """Fastq2bcl main function Tests with UMI"""
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        str(tmpdir), "data/test/03_single_with_umi/single_with_umi.fastq.gz"
    )
    assert seqdesc_fields["UMI"] == "ACGTAGTAC"


def test_fastq2bcl_with_exclude_umi(tmpdir):
    """Fastq2bcl main function Tests with exclude-umi"""
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        str(tmpdir),
        "data/test/03_single_with_umi/single_with_umi.fastq.gz",
        exclude_umi=True,
    )
    assert seqdesc_fields["UMI"] == "ACGTAGTAC"


def test_fastq2bcl_with_index(tmpdir):
    """Fastq2bcl main function Tests with INDEX"""
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        outdir=str(tmpdir),
        r1="data/test/02_single_with_index/single.R1.fastq.gz",
    )
    assert seqdesc_fields["index"] == "AACCACTA"


def test_fastq2bcl_with_exclude_index(tmpdir):
    """Fastq2bcl main function Tests with exclude-index"""
    run_id, rundir, seqdesc_fields, mask_string = fastq2bcl(
        outdir=str(tmpdir),
        r1="data/test/02_single_with_index/single.R1.fastq.gz",
        exclude_index=True,
    )
    assert seqdesc_fields["index"] == "AACCACTA"

tests/test_cli.py
tucano/fastq2bcl
import pytest

from fastq2bcl.parser import parse_seqdesc_fields, validate_fields

__author__ = "Davide Rambaldi"
__copyright__ = "Davide Rambaldi"
__license__ = "MIT"

expected_full = {
    "instrument": "M11111",
    "run_number": "222",
    "flowcell_id": "000000000-K9H97",
    "lane": "1",
    "tile": "1101",
    "x_pos": "19304",
    "y_pos": "1328",
    "UMI": "AAACGGG",
    "read": "1",
    "is_filtered": "N",
    "control_number": "0",
    "index": "1",
}

expected_no_umi = {
    "instrument": "M11111",
    "run_number": "222",
    "flowcell_id": "000000000-K9H97",
    "lane": "1",
    "tile": "1101",
    "x_pos": "19304",
    "y_pos": "1328",
    "UMI": None,
    "read": "1",
    "is_filtered": "N",
    "control_number": "0",
    "index": "1",
}

expected_missing = {
    "instrument": "M11111",
    "run_number": "222",
    "flowcell_id": "000000000-K9H97",
    "lane": "1",
    "tile": "1101",
    "x_pos": None,
    "y_pos": "1328",
    "UMI": None,
    "read": "1",
    "is_filtered": "N",
    "control_number": "0",
    "index": "1",
}


def test_parse_umi():
    """Parse UMI field tests"""
    fields = parse_seqdesc_fields(
        "M11111:222:000000000-K9H97:1:1101:19304:1328:AAACGGG 1:N:0:1"
    )
    assert fields["UMI"] == "AAACGGG"


def test_parse_seqdesc_fields():
    """SeqIO description parser Tests"""
    assert (
        parse_seqdesc_fields("M11111:222:000000000-K9H97:1:1101:19304:1328 1:N:0:1")
        == expected_no_umi
    )
    assert (
        parse_seqdesc_fields(
            "M11111:222:000000000-K9H97:1:1101:19304:1328:AAACGGG 1:N:0:1"
        )
        == expected_full
    )
    with pytest.raises(ValueError):
        parse_seqdesc_fields("AAA:1:2")
        parse_seqdesc_fields("M11111:222:000000000-K9H97:1:1101:::AAACGGG 1:N:0:1")


def test_validate_fields():
    """SeqIO fields validationTests"""
    assert validate_fields(expected_full) == expected_full
    assert validate_fields(expected_no_umi) == expected_no_umi
    with pytest.raises(ValueError):
        validate_fields(expected_missing)

tests/test_parser.py
tucano/fastq2bcl
import pytest

from fastq2bcl.reader import (
    read_first_record,
    read_fastq_files,
    get_mask_from_files,
)

__author__ = "Davide Rambaldi"
__copyright__ = "Davide Rambaldi"
__license__ = "MIT"

expected_data_single_seq = [
    (
        "CTTCCTAGAAGTACGTGCCAGCACGATCCAATCTCGCATCACCTTTTTTCTTTCTACTTCTACTCTCCTCTTATCTCTTCTTTTTCTTGTTTTTTTTCTTTATTCCATCT",
        [
            34,
            34,
            34,
            34,
            34,
            37,
            32,
            11,
            11,
            11,
            11,
            34,
            24,
            34,
            21,
            36,
            12,
            25,
            34,
            24,
            11,
            34,
            11,
            34,
            10,
            11,
            25,
            36,
            34,
            24,
            11,
            34,
            37,
            35,
            36,
            11,
            31,
            10,
            21,
            10,
            26,
            11,
            11,
            11,
            34,
            11,
            34,
            37,
            22,
            11,
            31,
            24,
            36,
            11,
            11,
            11,
            34,
            27,
            11,
            11,
            11,
            26,
            27,
            34,
            11,
            11,
            21,
            11,
            11,
            25,
            34,
            31,
            11,
            11,
            11,
            11,
            25,
            27,
            27,
            31,
            11,
            11,
            11,
            20,
            28,
            32,
            11,
            27,
            11,
            11,
            11,
            19,
            11,
            24,
            28,
            25,
            31,
            27,
            30,
            11,
            11,
            11,
            11,
            24,
            34,
            11,
            11,
            24,
            11,
            11,
        ],
    )
]

expected_data_single_pos = [("21270", "1316")]

expected_data_pair_seq = [
    (
        "ACTGAACCACTACTGAGCTGGGACGGAGTATACATTAACATAAACGTATCATAGCTTAGGACACTGCTCATAGCTGAGACGGATGTCATGACTGAGTTAGGGCACCTGGGATTAGTGTTTAACACTCTTTTGATACGGTATTGCCTAGATGGTACCCCAGCCGCGCTCCGGAGTCCAATAAAGCAGGGAAATACGGTCTACAGAGTGAGCCGACGCACGCAGTCACGAATCAGCCGGGCAGCTCATGGTACTGAGGTGACCATTCCCCTTTACCCCGTAACCGTCTCTTGTCTCCCGTGGCCCGATCGAGAACTTATCTCCAAGAAGCAAACGTTGAAGCACTTGTACGCAGCAGACACCTGCTAAGCACAATAACCGCGTCGGTGCGCCTTGGGGTTACAAATAGGAAGTAGTAGAACGTTAGATAATCCCTCTCAAACAACTTTAGTACCTTCCTAAACGTTTACCCGTATAATGTGTTGCTCGCACTAACTTCTGTCGTTTGCACGAATGTATTTTCGATCGGGCCACGGGAGACAAGAGACGGTTACGGGGTAAAGGGGAATGGTCACCTCAGTACCATGAGCTGCCCGGCTGATTCGTGACTGCGTGCGTCGG",
        [40] * 618,
    )
]

expected_data_multi_samples_seq_1 = "ACTGAACCACTACTGAGCTGGGACGGAGTATACATTAACATAAACGTATCATAGCTTAGGACACTGCTCATAGCTGAGACGGATGTCATGACTGAGTTAGGGCACCTGGGATTAGTGTTTAACACTCTTTTGATACGGTATTGCCTAGATGGTACCCCAGCCGCGCTCCGGAGTCCAATAAAGCAGGGAAATACGGTCTACAGAGTGAGCCGACGCACGCAGTCACGAATCAGCCGGGCAGCTCATGGTACTGAGGTGACCATTCCCCTTTACCCCGTAACCGTCTCTTGTCTCCCAACCACTAAACCACTAGAACTTATCTCCAAGAAGCAAACGTTGAAGCACTTGTACGCAGCAGACACCTGCTAAGCACAATAACCGCGTCGGTGCGCCTTGGGGTTACAAATAGGAAGTAGTAGAACGTTAGATAATCCCTCTCAAACAACTTTAGTACCTTCCTAAACGTTTACCCGTATAATGTGTTGCTCGCACTAACTTCTGTCGTTTGCACGAATGTATTTTCGATCGGGCCACGGGAGACAAGAGACGGTTACGGGGTAAAGGGGAATGGTCACCTCAGTACCATGAGCTGCCCGGCTGATTCGTGACTGCGTGCGTCGG"

expected_data_pair_pos = [("1", "2")]
expected_data_multi_samples_pos = [("1", "2"), ("3", "4")]


def test_read_first_record():
    r = read_first_record("data/test/01_single/test_single.fastq.gz")
    assert (
        str(r.seq)
        == "CTTCCTAGAAGTACGTGCCAGCACGATCCAATCTCGCATCACCTTTTTTCTTTCTACTTCTACTCTCCTCTTATCTCTTCTTTTTCTTGTTTTTTTTCTTTATTCCATCT"
    )


def test_read_single_fastq_files():
    seq, pos = read_fastq_files(
        "data/test/01_single/test_single.fastq.gz", None, None, None, True, True
    )
    assert seq == expected_data_single_seq
    assert pos == expected_data_single_pos


def test_read_pair_fastq_files():
    seq, pos = read_fastq_files(
        "data/test/07_pair/R1.fastq.gz",
        "data/test/07_pair/R2.fastq.gz",
        None,
        None,
        True,
        True,
    )
    assert seq == expected_data_pair_seq
    assert pos == expected_data_pair_pos


def test_read_pair_with_double_index():
    seq, pos = read_fastq_files(
        "data/test/05_multi_pair_double_index/R1.fastq.gz",
        "data/test/05_multi_pair_double_index/R2.fastq.gz",
        "data/test/05_multi_pair_double_index/RIndex1.fastq.gz",
        "data/test/05_multi_pair_double_index/RIndex2.fastq.gz",
        True,
        True,
    )
    assert seq[0][0] == expected_data_multi_samples_seq_1
    assert pos == expected_data_multi_samples_pos


def test_get_mask_from_files():
    assert (
        get_mask_from_files(
            "data/test/05_multi_pair_double_index/R1.fastq.gz",
            "data/test/05_multi_pair_double_index/R2.fastq.gz",
            "data/test/05_multi_pair_double_index/RIndex1.fastq.gz",
            "data/test/05_multi_pair_double_index/RIndex2.fastq.gz",
            True,
            True,
        )
        == "296N8Y8Y309N"
    )


def test_get_mask_from_files_raise_umi_error():
    with pytest.raises(ValueError):
        get_mask_from_files(
            "data/test/03_single_with_umi/single_with_umi.fastq.gz",
            None,
            "data/test/05_multi_pair_double_index/RIndex1.fastq.gz",
            "data/test/05_multi_pair_double_index/RIndex2.fastq.gz",
            False,
            True,
        )


def test_get_mask_from_files_raise_index_error():
    with pytest.raises(ValueError):
        get_mask_from_files(
            "data/test/05_multi_pair_double_index/R1.fastq.gz",
            "data/test/05_multi_pair_double_index/R2.fastq.gz",
            "data/test/05_multi_pair_double_index/RIndex1.fastq.gz",
            "data/test/05_multi_pair_double_index/RIndex2.fastq.gz",
            True,
            False,
        )


def test_get_mask_from_file_with_umi():
    assert (
        get_mask_from_files(
            "data/test/03_single_with_umi/single_with_umi.fastq.gz",
            None,
            None,
            None,
            False,
            True,
        )
        == "110N9Y"
    )


def test_read_fastq_file_with_umi():
    seq, pos = read_fastq_files(
        "data/test/03_single_with_umi/single_with_umi.fastq.gz",
        None,
        None,
        None,
        False,
        True,
    )
    assert seq[0][0].endswith("ACGTAGTAC") == True


def test_seq_mismatch():
    with pytest.raises(ValueError):
        seq, pos = read_fastq_files(
            "data/test/01_single/test_single.fastq.gz",
            "data/test/07_pair/R2.fastq.gz",
            None,
            None,
            True,
            True,
        )

tests/test_reader.py
tucano/fastq2bcl
import pytest
import struct

from fastq2bcl.writer import (
    write_run_info_xml,
    generate_run_info_xml,
    write_filter,
    write_control,
    encode_loc_bytes,
    write_locs,
    encode_cluster_byte,
    write_bcls_and_stats,
)

__author__ = "Davide Rambaldi"
__copyright__ = "Davide Rambaldi"
__license__ = "MIT"

excepted_xml = """<?xml version="1.0"?>
<RunInfo xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" Version="2">
    <Run Id="YYMMDD_M11111_0222_000000000-K9H97" Number="222">
        <Flowcell>000000000-K9H97</Flowcell>
        <Instrument>M11111</Instrument>
        <Date>YYMMDD</Date>
        <Reads>
            <Read NumCycles="110" Number="1" IsIndexedRead="N" />
        </Reads>
        <FlowcellLayout LaneCount="1" SurfaceCount="1" SwathCount="1" TileCount="1" />
    </Run>
</RunInfo>
"""

expected_filter = b"\x00\x00\x00\x00\x03\x00\x00\x00\x01\x00\x00\x00\x01"
expected_control = b"\x00\x00\x00\x00\x02\x00\x00\x00\x01\x00\x00\x00\x00\x00"
expected_locs = (
    b"\x01\x00\x00\x00\x00\x00\x80?\x01\x00\x00\x00\xcd\xcc\xc7\xc2\xcd\xcc\xc7\xc2"
)
expected_bcl = b"\x01\x00\x00\x00\x05"
expected_stats = b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"

test_sequences = [(["C"], [1])]

test_mask = [{"cycles": 110, "index": "N", "id": 1}]


def test_generate_run_info_xml():
    assert (
        generate_run_info_xml(
            "YYMMDD_M11111_0222_000000000-K9H97",
            222,
            "000000000-K9H97",
            "M11111",
            test_mask,
        )
        == excepted_xml
    )


def test_write_run_info_xml(tmp_path):
    xmlout = tmp_path / "RunInfo.xml"
    write_run_info_xml(
        tmp_path,
        "YYMMDD_M11111_0222_000000000-K9H97",
        222,
        "000000000-K9H97",
        "M11111",
        test_mask,
    )
    assert xmlout.read_text() == excepted_xml


def test_encode_loc_bytes():
    assert encode_loc_bytes(1, 1) == b"\xcd\xcc\xc7\xc2\xcd\xcc\xc7\xc2"


def test_write_filter(tmp_path):
    binaryout = tmp_path / "Data/Intensities/BaseCalls/L001/s_1_1101.filter"
    write_filter(tmp_path, 1)
    with open(binaryout, "rb") as binfile:
        binary_content = binfile.read()
        assert binary_content == expected_filter


def test_write_control(tmp_path):
    binaryout = tmp_path / "Data/Intensities/BaseCalls/L001/s_1_1101.control"
    write_control(tmp_path, 1)
    with open(binaryout, "rb") as binfile:
        binary_content = binfile.read()
        assert binary_content == expected_control


def test_write_locs(tmp_path):
    binaryout = tmp_path / "Data/Intensities/L001/s_1_1101.locs"
    write_locs(tmp_path, [(1, 1)])
    with open(binaryout, "rb") as binfile:
        binary_content = binfile.read()
        assert binary_content == expected_locs


def test_encode_cluster_byte():
    assert encode_cluster_byte("A", 1) == b"\x04"


def test_encode_cluster_byte_null():
    assert encode_cluster_byte("N", 1) == b"\x00"


def test_write_bcls_and_stats(tmp_path):
    binaryout = tmp_path / "Data/Intensities/BaseCalls/L001/C1.1/s_1_1101.bcl"
    statsout = tmp_path / "Data/Intensities/BaseCalls/L001/C1.1/s_1_1101.stats"
    write_bcls_and_stats(tmp_path, test_sequences)
    with open(binaryout, "rb") as binfile:
        binary_content = binfile.read()
        assert binary_content == expected_bcl
    with open(statsout, "rb") as binfile:
        binary_content = binfile.read()
        assert binary_content == expected_stats

tests/test_writer.py
tucano/fastq2bcl