16568275773

Committed 28 Jul 2025 11:51AM UTC coverage: 69.115% (-0.6%) from 69.722%

Build # 16568275773

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #870 from cokelaer/dev

add threshold parameter in G4 module

Run Details

1 of 2 new or added lines in 1 file covered. (50.0%)

725 existing lines in 17 files now uncovered.

14389 of 20819 relevant lines covered (69.11%)

2.07 hits per line

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

42.91

/sequana/modules_report/summary.py

#
#  This file is part of Sequana software
#
#  Copyright (c) 2016 - 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
#
##############################################################################
"""Module to write summary.html have all information about the pipeline and
to visit other analysis"""
import importlib.metadata
import os

from deprecated import deprecated

from sequana.lazy import pandas as pd
from sequana.modules_report.base_module import SequanaBaseModule
from sequana.utils import config
from sequana.utils.datatables_js import DataTable


class SummaryBase(SequanaBaseModule):
    def __init__(self, required_dir=None):
        super(SummaryBase, self).__init__(required_dir=required_dir)

    def dependencies(self):
        """Table with all python dependencies and a text file with tools
        needed and their versions.
        """
        html_table = self.get_table_dependencies()
        pypi = self.create_link("Pypi", "http://pypi.python.org")
        try:
            if "requirements" in self.json:
                req = self.copy_file(self.json["requirements"], "inputs")
            else:
                raise Exception
        except:
            try:
                req = self.json["requirements"]
            except:
                return
        req = self.create_link("requirements", req)
        content = "<p>Python dependencies (<b>{0}</b>){1}</p>".format(pypi, html_table, req)
        l, c = self.create_hide_section("Dep", "collapse/expand", content, hide=True)
        self.sections.append(
            {
                "name": "Dependencies {0}".format(self.add_float_right("<small>{0}</small>".format(l))),
                "anchor": "dependencies",
                "content": c,
            }
        )

    def get_table_dependencies(self):
        """Return dependencies of Sequana."""

        project_names = list()
        versions = list()
        links = list()
        pypi = "https://pypi.python.org/pypi/{0}"

        for dep in importlib.metadata.requires("sequana"):

            try:
                project_name, version = dep.split()
            except ValueError:
                project_name = dep
                version = "?"

            versions.append(version)
            project_names.append(project_name)
            links.append(pypi.format(project_name))

        df = pd.DataFrame({"package": project_names, "version": versions, "link": links})
        df["sort"] = df["package"].str.lower()
        df.sort_values(by="sort", axis=0, inplace=True)
        df.drop("sort", axis=1, inplace=True)
        datatable = DataTable(df, "dep")
        datatable.datatable.datatable_options = {
            "paging": "false",
            "bFilter": "false",
            "bInfo": "false",
            "bSort": "false",
        }
        datatable.datatable.set_links_to_column("link", "package")
        js = datatable.create_javascript_function()
        html = datatable.create_datatable()
        return js + "\n" + html


@deprecated(version="1.0", reason="will be removed in v1.0. Update your pipelines to use SequanaReport instead")
class SummaryModule(SummaryBase):
    """Write summary HTML report of an analysis. It contains all information
    about the pipeline used, input/output files and version of software.
    """

    def __init__(self, data, intro="", output_filename="summary.html"):
        """ """
        super().__init__()
        print("SummaryModule will be deprecated in the future. Use SummaryModule2 for now")
        self.json = data
        for this in ["inputs", "outputs"]:
            assert this in self.json
        self.title = "Sequana Report Summary"
        self.intro = intro
        self.create_report_content()
        self.create_html(output_filename)

    def create_report_content(self):
        """Create the report content."""
        self.sections = list()

        if self.json["inputs"]:
            self.pipeline_inputs()
        if self.json["outputs"]:
            self.pipeline_outputs()
        if self.json["html"]:
            self.pipeline_html()

        for section in config.summary_sections:
            self.sections.append(section)

        self.workflow()
        self.dependencies()

    def pipeline_inputs(self):
        """Links corresponding to the analysed input files."""
        # copy inputs in the input directory
        input_dir = "inputs"

        inputs = [self.copy_file(i, input_dir) for i in self.json["inputs"]]
        # create links list
        html_list = "<li>{0}</li>"
        links = [
            html_list.format(self.create_link(os.path.basename(i), i, newtab=False, download=True)) for i in inputs
        ]
        links = "<ul>{0}</ul>".format("\n".join(links))
        self.sections.append(
            {
                "name": "Inputs",
                "anchor": "input",
                "content": "<p>Link to the original data analysed.</p>\n" "{0}".format(links),
            }
        )

    def pipeline_outputs(self):
        """Links to important outputs generated by the pipeline"""
        # copy outputs in the output directory
        output_dir = "outputs"
        outputs = [self.copy_file(i, output_dir) for i in self.json["outputs"]]
        # create links list
        html_list = "<li>{0}</li>"
        links = [
            html_list.format(self.create_link(os.path.basename(i), i, newtab=False, download=True)) for i in outputs
        ]
        links = "<ul>{0}</ul>".format("\n".join(links))
        self.sections.append(
            {
                "name": "Outputs",
                "anchor": "outputs",
                "content": "<p>Link to the most important output files generated by the "
                "pipeline.</p>\n{0}".format(links),
            }
        )

    def pipeline_html(self):
        """Links to HTML pages created by the rules."""
        output_dir = "html"
        html = [self.copy_file(i, output_dir) for i in self.json["html"]]
        html_list = "<li>{0}</li>"
        links = [html_list.format(self.create_link(os.path.basename(i), i)) for i in html]
        links = "<ul>{0}</ul>".format("\n".join(links))
        self.sections.append(
            {
                "name": "External HTML",
                "anchor": "ext_html",
                "content": "<p>Link to HTML pages created by the pipeline.</p>\n{0}" "\n".format(links),
            }
        )

    def workflow(self):
        """Create the interactive DAG to navigate through pages."""
        snakefile = self.copy_file(self.json["snakefile"], "./inputs")
        configfile = self.copy_file(self.json["config"], "./inputs")
        # move the SVG file in the images directory
        img = self.copy_file(self.json["rulegraph"], "./images")
        dag_svg = self.include_svg_image(img, alt="rulegraph")
        with open(self.json["snakefile"], "r") as fp:
            code = self.add_code_section(fp.read(), "python")
        sf = self.create_hide_section("Sf", "Show/hide Snakemake file", code, hide=True)
        sf = "\n".join(sf)
        with open(self.json["config"], "r") as fp:
            code = self.add_code_section(fp.read(), "yaml")
        c = self.create_hide_section("C", "Show/hide config file", code, hide=True)
        c = "\n".join(c)
        self.sections.append(
            {
                "name": "Workflow",
                "anchor": "workflow",
                "content": "<p>The following network shows the workflow of the pipeline. "
                "Blue boxes are clickable and redirect to dedicated reports."
                "</p>\n{0}\n"
                "<p>The analysis was performed with the following "
                '<a href="{3}">Snakemake</a> and <a href="{4}">configfile</a>:'
                "</p>\n"
                "<ul>\n"
                "    <li>{1}</li>\n"
                "    <li>{2}</li>\n"
                "</ul>".format(dag_svg, sf, c, snakefile, configfile),
            }
        )


@deprecated(version="1.0", reason="will be removed in v1.0. Update your pipelines to use SequanaReport instead")
class SummaryModule2(SummaryBase):
    """Write summary HTML report of an analysis. It contains all information
    about the pipeline used, input/output files and version of software.
    """

    def __init__(
        self,
        data,
        intro="",
        output_filename="summary.html",
        title="",
        workflow=True,
        Nsamples=1,
    ):
        """ """
        super(SummaryModule2, self).__init__(required_dir=("js", "css"))
        self.json = data
        self.name = data.get("name", "undefined")
        self.title = f"Sequana Report Summary ({self.name})"
        self.wrappers = data.get("sequana_wrappers", "latest")
        self.intro = intro

        config.pipeline_version = data.get("pipeline_version", "latest")
        config.pipeline_name = data.get("name", "undefined")
        config.sequana_wrappers = self.wrappers

        self.create_report_content(workflow=workflow)
        self.create_html(output_filename)

    def create_report_content(self, workflow=True):
        """Create the report content."""
        self.sections = list()

        for section in config.summary_sections:
            self.sections.append(section)

        if workflow:
            self.workflow()
        self.dependencies()
        self.caller()

    def workflow(self):
        img = self.json["rulegraph"]
        dag_svg = self.include_svg_image(img, alt="workflow")
        snakefile = ".sequana/{}.rules".format(self.name)

        try:
            with open(snakefile, "r") as fp:
                code = self.add_code_section(fp.read(), "python")
            sf = self.create_hide_section("Sf", "Show/hide Snakemake file", code, hide=True)
            sf = "\n".join(sf)
        except IOError:
            sf = "no snakefile found in .sequana/"

        configfile = ".sequana/config.yaml"
        try:
            with open(configfile, "r") as fp:
                code = self.add_code_section(fp.read(), "yaml")
                c = self.create_hide_section("C", "Show/hide config file", code, hide=True)
                c = "\n".join(c)
        except IOError:
            c = "no config found in .sequana/"

        self.sections.append(
            {
                "name": "Workflow",
                "anchor": "workflow",
                "content": "<p>The following network shows the workflow of the pipeline. "
                "Blue boxes are clickable and redirect to dedicated reports."
                "</p>\n{0}\n"
                "<p>The analysis was performed with the following "
                '<a href="{3}">Snakemake</a> and <a href="{4}">configfile</a>:'
                "</p>\n"
                "<ul>\n"
                "    <li>{1}</li>\n"
                "    <li>{2}</li>\n"
                "</ul>".format(dag_svg, sf, c, snakefile, configfile),
            }
        )

    def get_table_versions(self):
        """Return third party tools from the requirements.txt and their versions."""

        # if no version.txt is found, return nothing
        if os.path.exists(".sequana/versions.txt") is False:
            return ""

        versions = []
        tools = []
        with open(".sequana/versions.txt", "r") as fin:
            for line in fin.readlines():
                tool, version = line.split()
                versions.append(version)
                tools.append(tool)

        df = pd.DataFrame({"tool": tools, "version": versions})
        df["sort"] = df["tool"].str.lower()
        df.sort_values(by="sort", axis=0, inplace=True)
        df.drop("sort", axis=1, inplace=True)
        datatable = DataTable(df, "dep_and_version")
        datatable.datatable.datatable_options = {
            "paging": "false",
            "bFilter": "false",
            "bInfo": "false",
            "bSort": "false",
        }
        js = datatable.create_javascript_function()
        html = datatable.create_datatable()
        return js + "\n" + html

    def dependencies(self):
        """Table with all python dependencies and a text file with tools
        needed and their versions.
        """

        html_table_versions = self.get_table_versions()
        html_table_deps = self.get_table_dependencies()
        pypi = self.create_link("Pypi", "http://pypi.python.org")

        req = self.create_link("requirements", ".sequana/env.yml")

        content = (
            "<p>Third party tools can be found within containers (see config file) if you use --use-apptainers options. Otherwise, here is a list of required dependencies and their versions.</p>"
            "<p>{3}</p>"
            "<p>Python dependencies (<b>{0}</b>){1}</p>".format(pypi, html_table_deps, req, html_table_versions)
        )
        l, c = self.create_hide_section("Dep", "collapse/expand", content, hide=True)
        self.sections.append(
            {
                "name": "Dependencies {0}".format(self.add_float_right("<small>{0}</small>".format(l))),
                "anchor": "dependencies",
                "content": c,
            }
        )

    def caller(self):
        try:
            with open(".sequana/info.txt") as fin:
                command = fin.readlines()
                command = "<pre>" + "\n".join([x for x in command if not x.startswith("#")]) + "</pre>"
        except Exception as err:
            print(err)
            command = "unknown"
        self.sections.append({"name": "Command", "anchor": "command", "content": command})


class SequanaReport(SummaryBase):
    """Write summary HTML report of an analysis. It contains all information
    about the pipeline used, input/output files and version of software.
    """

    def __init__(
        self,
        data,
        intro="",
        output_filename="summary.html",
        title="",
        workflow=True,
        Nsamples=1,
    ):
        """ """
        super(SequanaReport, self).__init__(required_dir=("js", "css"))
        self.json = data
        self.name = data.get("name", "undefined")
        self.title = f"Sequana Report Summary ({self.name})"
        self.wrappers = data.get("sequana_wrappers", "latest")
        self.intro = intro

        config.pipeline_version = data.get("pipeline_version", "latest")
        config.pipeline_name = data.get("name", "undefined")
        config.sequana_wrappers = self.wrappers

        self.create_report_content(workflow=workflow)
        self.create_html(output_filename)

    def create_report_content(self, workflow=True):
        """Create the report content."""
        self.sections = list()

        for section in config.summary_sections:
            self.sections.append(section)

        if workflow:
            self.workflow()
        self.dependencies()
        self.caller()

    def workflow(self):
        img = self.json["rulegraph"]
        dag_svg = self.include_svg_image(img, alt="workflow")
        snakefile = ".sequana/{}.rules".format(self.name)

        try:
            with open(snakefile, "r") as fp:
                code = self.add_code_section(fp.read(), "python")
            sf = self.create_hide_section("Sf", "Show/hide Snakemake file", code, hide=True)
            sf = "\n".join(sf)
        except IOError:
            sf = "no snakefile found in .sequana/"

        configfile = ".sequana/config.yaml"
        try:
            with open(configfile, "r") as fp:
                code = self.add_code_section(fp.read(), "yaml")
                c = self.create_hide_section("C", "Show/hide config file", code, hide=True)
                c = "\n".join(c)
        except IOError:
            c = "no config found in .sequana/"

        self.sections.append(
            {
                "name": "Workflow",
                "anchor": "workflow",
                "content": "<p>The following network shows the workflow of the pipeline. "
                "Blue boxes are clickable and redirect to dedicated reports."
                "</p>\n{0}\n"
                "<p>The analysis was performed with the following "
                '<a href="{3}">Snakemake</a> and <a href="{4}">configfile</a>:'
                "</p>\n"
                "<ul>\n"
                "    <li>{1}</li>\n"
                "    <li>{2}</li>\n"
                "</ul>".format(dag_svg, sf, c, snakefile, configfile),
            }
        )

    def get_table_versions(self):
        """Return third party tools from the requirements.txt and their versions."""

        # if no version.txt is found, return nothing
        if os.path.exists(".sequana/versions.txt") is False:
            return ""

        versions = []
        tools = []
        with open(".sequana/versions.txt", "r") as fin:
            for line in fin.readlines():
                tool, version = line.split()
                versions.append(version)
                tools.append(tool)

        df = pd.DataFrame({"tool": tools, "version": versions})
        try:
            df["sort"] = df["tool"].str.lower()
            df.sort_values(by="sort", axis=0, inplace=True)
            df.drop("sort", axis=1, inplace=True)
        except (KeyError, AttributeError):  # could be empty
            pass
        datatable = DataTable(df, "dep_and_version")
        datatable.datatable.datatable_options = {
            "paging": "false",
            "bFilter": "false",
            "bInfo": "false",
            "bSort": "false",
        }
        js = datatable.create_javascript_function()
        html = datatable.create_datatable()
        return js + "\n" + html

    def dependencies(self):
        """Table with all python dependencies and a text file with tools
        needed and their versions.
        """

        html_table_versions = self.get_table_versions()
        html_table_deps = self.get_table_dependencies()
        pypi = self.create_link("Pypi", "http://pypi.python.org")

        req = self.create_link("requirements", ".sequana/env.yml")

        content = (
            "<p>Third party tools can be found within containers (see config file abobe) if you use --use-apptainers option. Otherwise, here is a list of required dependencies and their versions.</p>"
            "<p>{3}</p>"
            "<p>Python dependencies (<b>{0}</b>){1}</p>".format(pypi, html_table_deps, req, html_table_versions)
        )
        l, c = self.create_hide_section("Dep", "collapse/expand", content, hide=True)
        self.sections.append(
            {
                "name": "Dependencies {0}".format(self.add_float_right("<small>{0}</small>".format(l))),
                "anchor": "dependencies",
                "content": c,
            }
        )

    def caller(self):
        try:
            with open(".sequana/info.txt") as fin:
                command = fin.readlines()
                command = "<pre>" + "\n".join([x for x in command if not x.startswith("#")]) + "</pre>"
        except Exception as err:
            print(err)
            command = "unknown"
        self.sections.append({"name": "Command", "anchor": "command", "content": command})

1	#
2	# This file is part of Sequana software
3	#
4	# Copyright (c) 2016 - Sequana Development Team
5	#
6	#
7	# Distributed under the terms of the 3-clause BSD license.
8	# The full license is in the LICENSE file, distributed with this software.
9	#
10	# website: https://github.com/sequana/sequana
11	# documentation: http://sequana.readthedocs.io
12	#
13	##############################################################################
14	"""Module to write summary.html have all information about the pipeline and	3✔
15	to visit other analysis"""
16	import importlib.metadata	3✔
17	import os	3✔
18
19	from deprecated import deprecated	3✔
20
21	from sequana.lazy import pandas as pd	3✔
22	from sequana.modules_report.base_module import SequanaBaseModule	3✔
23	from sequana.utils import config	3✔
24	from sequana.utils.datatables_js import DataTable	3✔
25
26
27	class SummaryBase(SequanaBaseModule):	3✔
28	def __init__(self, required_dir=None):	3✔
29	super(SummaryBase, self).__init__(required_dir=required_dir)	3✔
30
31	def dependencies(self):	3✔
32	"""Table with all python dependencies and a text file with tools
33	needed and their versions.
34	"""
35	html_table = self.get_table_dependencies()	3✔
36	pypi = self.create_link("Pypi", "http://pypi.python.org")	3✔
37	try:	3✔
38	if "requirements" in self.json:	3✔
39	req = self.copy_file(self.json["requirements"], "inputs")	3✔
40	else:
41	raise Exception	×
42	except:	×
43	try:	×
44	req = self.json["requirements"]	×
45	except:	×
46	return	×
47	req = self.create_link("requirements", req)	3✔
48	content = "<p>Python dependencies (<b>{0}</b>){1}</p>".format(pypi, html_table, req)	3✔
49	l, c = self.create_hide_section("Dep", "collapse/expand", content, hide=True)	3✔
50	self.sections.append(	3✔
51	{
52	"name": "Dependencies {0}".format(self.add_float_right("<small>{0}</small>".format(l))),
53	"anchor": "dependencies",
54	"content": c,
55	}
56	)
57
58	def get_table_dependencies(self):	3✔
59	"""Return dependencies of Sequana."""
60
61	project_names = list()	3✔
62	versions = list()	3✔
63	links = list()	3✔
64	pypi = "https://pypi.python.org/pypi/{0}"	3✔
65
66	for dep in importlib.metadata.requires("sequana"):	3✔
67
68	try:	3✔
69	project_name, version = dep.split()	3✔
UNCOV 70	except ValueError:	×
UNCOV 71	project_name = dep	×
UNCOV 72	version = "?"	×
73
74	versions.append(version)	3✔
75	project_names.append(project_name)	3✔
76	links.append(pypi.format(project_name))	3✔
77
78	df = pd.DataFrame({"package": project_names, "version": versions, "link": links})	3✔
79	df["sort"] = df["package"].str.lower()	3✔
80	df.sort_values(by="sort", axis=0, inplace=True)	3✔
81	df.drop("sort", axis=1, inplace=True)	3✔
82	datatable = DataTable(df, "dep")	3✔
83	datatable.datatable.datatable_options = {	3✔
84	"paging": "false",
85	"bFilter": "false",
86	"bInfo": "false",
87	"bSort": "false",
88	}
89	datatable.datatable.set_links_to_column("link", "package")	3✔
90	js = datatable.create_javascript_function()	3✔
91	html = datatable.create_datatable()	3✔
92	return js + "\n" + html	3✔
93
94
95	@deprecated(version="1.0", reason="will be removed in v1.0. Update your pipelines to use SequanaReport instead")	3✔
96	class SummaryModule(SummaryBase):	3✔
97	"""Write summary HTML report of an analysis. It contains all information
98	about the pipeline used, input/output files and version of software.
99	"""
100
101	def __init__(self, data, intro="", output_filename="summary.html"):	3✔
102	""" """
103	super().__init__()	3✔
104	print("SummaryModule will be deprecated in the future. Use SummaryModule2 for now")	3✔
105	self.json = data	3✔
106	for this in ["inputs", "outputs"]:	3✔
107	assert this in self.json	3✔
108	self.title = "Sequana Report Summary"	3✔
109	self.intro = intro	3✔
110	self.create_report_content()	3✔
111	self.create_html(output_filename)	3✔
112
113	def create_report_content(self):	3✔
114	"""Create the report content."""
115	self.sections = list()	3✔
116
117	if self.json["inputs"]:	3✔
118	self.pipeline_inputs()	3✔
119	if self.json["outputs"]:	3✔
120	self.pipeline_outputs()	3✔
121	if self.json["html"]:	3✔
122	self.pipeline_html()	3✔
123
124	for section in config.summary_sections:	3✔
UNCOV 125	self.sections.append(section)	×
126
127	self.workflow()	3✔
128	self.dependencies()	3✔
129
130	def pipeline_inputs(self):	3✔
131	"""Links corresponding to the analysed input files."""
132	# copy inputs in the input directory
133	input_dir = "inputs"	3✔
134
135	inputs = [self.copy_file(i, input_dir) for i in self.json["inputs"]]	3✔
136	# create links list
137	html_list = "<li>{0}</li>"	3✔
138	links = [	3✔
139	html_list.format(self.create_link(os.path.basename(i), i, newtab=False, download=True)) for i in inputs
140	]
141	links = "<ul>{0}</ul>".format("\n".join(links))	3✔
142	self.sections.append(	3✔
143	{
144	"name": "Inputs",
145	"anchor": "input",
146	"content": "<p>Link to the original data analysed.</p>\n" "{0}".format(links),
147	}
148	)
149
150	def pipeline_outputs(self):	3✔
151	"""Links to important outputs generated by the pipeline"""
152	# copy outputs in the output directory
153	output_dir = "outputs"	3✔
154	outputs = [self.copy_file(i, output_dir) for i in self.json["outputs"]]	3✔
155	# create links list
156	html_list = "<li>{0}</li>"	3✔
157	links = [	3✔
158	html_list.format(self.create_link(os.path.basename(i), i, newtab=False, download=True)) for i in outputs
159	]
160	links = "<ul>{0}</ul>".format("\n".join(links))	3✔
161	self.sections.append(	3✔
162	{
163	"name": "Outputs",
164	"anchor": "outputs",
165	"content": "<p>Link to the most important output files generated by the "
166	"pipeline.</p>\n{0}".format(links),
167	}
168	)
169
170	def pipeline_html(self):	3✔
171	"""Links to HTML pages created by the rules."""
172	output_dir = "html"	3✔
173	html = [self.copy_file(i, output_dir) for i in self.json["html"]]	3✔
174	html_list = "<li>{0}</li>"	3✔
175	links = [html_list.format(self.create_link(os.path.basename(i), i)) for i in html]	3✔
176	links = "<ul>{0}</ul>".format("\n".join(links))	3✔
177	self.sections.append(	3✔
178	{
179	"name": "External HTML",
180	"anchor": "ext_html",
181	"content": "<p>Link to HTML pages created by the pipeline.</p>\n{0}" "\n".format(links),
182	}
183	)
184
185	def workflow(self):	3✔
186	"""Create the interactive DAG to navigate through pages."""
187	snakefile = self.copy_file(self.json["snakefile"], "./inputs")	3✔
188	configfile = self.copy_file(self.json["config"], "./inputs")	3✔
189	# move the SVG file in the images directory
190	img = self.copy_file(self.json["rulegraph"], "./images")	3✔
191	dag_svg = self.include_svg_image(img, alt="rulegraph")	3✔
192	with open(self.json["snakefile"], "r") as fp:	3✔
193	code = self.add_code_section(fp.read(), "python")	3✔
194	sf = self.create_hide_section("Sf", "Show/hide Snakemake file", code, hide=True)	3✔
195	sf = "\n".join(sf)	3✔
196	with open(self.json["config"], "r") as fp:	3✔
197	code = self.add_code_section(fp.read(), "yaml")	3✔
198	c = self.create_hide_section("C", "Show/hide config file", code, hide=True)	3✔
199	c = "\n".join(c)	3✔
200	self.sections.append(	3✔
201	{
202	"name": "Workflow",
203	"anchor": "workflow",
204	"content": "<p>The following network shows the workflow of the pipeline. "
205	"Blue boxes are clickable and redirect to dedicated reports."
206	"</p>\n{0}\n"
207	"<p>The analysis was performed with the following "
208	'<a href="{3}">Snakemake</a> and <a href="{4}">configfile</a>:'
209	"</p>\n"
210	"<ul>\n"
211	" <li>{1}</li>\n"
212	" <li>{2}</li>\n"
213	"</ul>".format(dag_svg, sf, c, snakefile, configfile),
214	}
215	)
216
217
218	@deprecated(version="1.0", reason="will be removed in v1.0. Update your pipelines to use SequanaReport instead")	3✔
219	class SummaryModule2(SummaryBase):	3✔
220	"""Write summary HTML report of an analysis. It contains all information
221	about the pipeline used, input/output files and version of software.
222	"""
223
224	def __init__(	3✔
225	self,
226	data,
227	intro="",
228	output_filename="summary.html",
229	title="",
230	workflow=True,
231	Nsamples=1,
232	):
233	""" """
UNCOV 234	super(SummaryModule2, self).__init__(required_dir=("js", "css"))	×
UNCOV 235	self.json = data	×
UNCOV 236	self.name = data.get("name", "undefined")	×
UNCOV 237	self.title = f"Sequana Report Summary ({self.name})"	×
238	self.wrappers = data.get("sequana_wrappers", "latest")	×
239	self.intro = intro	×
240
241	config.pipeline_version = data.get("pipeline_version", "latest")	×
242	config.pipeline_name = data.get("name", "undefined")	×
243	config.sequana_wrappers = self.wrappers	×
244
245	self.create_report_content(workflow=workflow)	×
246	self.create_html(output_filename)	×
247
248	def create_report_content(self, workflow=True):	3✔
249	"""Create the report content."""
250	self.sections = list()	×
251
UNCOV 252	for section in config.summary_sections:	×
UNCOV 253	self.sections.append(section)	×
254
UNCOV 255	if workflow:	×
256	self.workflow()	×
257	self.dependencies()	×
UNCOV 258	self.caller()	×
259
260	def workflow(self):	3✔
261	img = self.json["rulegraph"]	×
262	dag_svg = self.include_svg_image(img, alt="workflow")	×
UNCOV 263	snakefile = ".sequana/{}.rules".format(self.name)	×
264
265	try:	×
266	with open(snakefile, "r") as fp:	×
267	code = self.add_code_section(fp.read(), "python")	×
UNCOV 268	sf = self.create_hide_section("Sf", "Show/hide Snakemake file", code, hide=True)	×
269	sf = "\n".join(sf)	×
270	except IOError:	×
271	sf = "no snakefile found in .sequana/"	×
272
273	configfile = ".sequana/config.yaml"	×
274	try:	×
275	with open(configfile, "r") as fp:	×
UNCOV 276	code = self.add_code_section(fp.read(), "yaml")	×
277	c = self.create_hide_section("C", "Show/hide config file", code, hide=True)	×
278	c = "\n".join(c)	×
279	except IOError:	×
280	c = "no config found in .sequana/"	×
281
282	self.sections.append(	×
283	{
284	"name": "Workflow",
285	"anchor": "workflow",
286	"content": "<p>The following network shows the workflow of the pipeline. "
287	"Blue boxes are clickable and redirect to dedicated reports."
288	"</p>\n{0}\n"
289	"<p>The analysis was performed with the following "
290	'<a href="{3}">Snakemake</a> and <a href="{4}">configfile</a>:'
291	"</p>\n"
292	"<ul>\n"
293	" <li>{1}</li>\n"
294	" <li>{2}</li>\n"
295	"</ul>".format(dag_svg, sf, c, snakefile, configfile),
296	}
297	)
298
299	def get_table_versions(self):	3✔
300	"""Return third party tools from the requirements.txt and their versions."""
301
302	# if no version.txt is found, return nothing
UNCOV 303	if os.path.exists(".sequana/versions.txt") is False:	×
UNCOV 304	return ""	×
305
UNCOV 306	versions = []	×
307	tools = []	×
308	with open(".sequana/versions.txt", "r") as fin:	×
UNCOV 309	for line in fin.readlines():	×
310	tool, version = line.split()	×
311	versions.append(version)	×
312	tools.append(tool)	×
313
314	df = pd.DataFrame({"tool": tools, "version": versions})	×
315	df["sort"] = df["tool"].str.lower()	×
316	df.sort_values(by="sort", axis=0, inplace=True)	×
UNCOV 317	df.drop("sort", axis=1, inplace=True)	×
318	datatable = DataTable(df, "dep_and_version")	×
319	datatable.datatable.datatable_options = {	×
320	"paging": "false",
321	"bFilter": "false",
322	"bInfo": "false",
323	"bSort": "false",
324	}
UNCOV 325	js = datatable.create_javascript_function()	×
UNCOV 326	html = datatable.create_datatable()	×
UNCOV 327	return js + "\n" + html	×
328
329	def dependencies(self):	3✔
330	"""Table with all python dependencies and a text file with tools
331	needed and their versions.
332	"""
333
UNCOV 334	html_table_versions = self.get_table_versions()	×
UNCOV 335	html_table_deps = self.get_table_dependencies()	×
UNCOV 336	pypi = self.create_link("Pypi", "http://pypi.python.org")	×
337
338	req = self.create_link("requirements", ".sequana/env.yml")	×
339
340	content = (	×
341	"<p>Third party tools can be found within containers (see config file) if you use --use-apptainers options. Otherwise, here is a list of required dependencies and their versions.</p>"
342	"<p>{3}</p>"
343	"<p>Python dependencies (<b>{0}</b>){1}</p>".format(pypi, html_table_deps, req, html_table_versions)
344	)
UNCOV 345	l, c = self.create_hide_section("Dep", "collapse/expand", content, hide=True)	×
UNCOV 346	self.sections.append(	×
347	{
348	"name": "Dependencies {0}".format(self.add_float_right("<small>{0}</small>".format(l))),
349	"anchor": "dependencies",
350	"content": c,
351	}
352	)
353
354	def caller(self):	3✔
UNCOV 355	try:	×
UNCOV 356	with open(".sequana/info.txt") as fin:	×
UNCOV 357	command = fin.readlines()	×
UNCOV 358	command = "<pre>" + "\n".join([x for x in command if not x.startswith("#")]) + "</pre>"	×
UNCOV 359	except Exception as err:	×
360	print(err)	×
361	command = "unknown"	×
362	self.sections.append({"name": "Command", "anchor": "command", "content": command})	×
363
364
365	class SequanaReport(SummaryBase):	3✔
366	"""Write summary HTML report of an analysis. It contains all information
367	about the pipeline used, input/output files and version of software.
368	"""
369
370	def __init__(	3✔
371	self,
372	data,
373	intro="",
374	output_filename="summary.html",
375	title="",
376	workflow=True,
377	Nsamples=1,
378	):
379	""" """
UNCOV 380	super(SequanaReport, self).__init__(required_dir=("js", "css"))	×
UNCOV 381	self.json = data	×
UNCOV 382	self.name = data.get("name", "undefined")	×
UNCOV 383	self.title = f"Sequana Report Summary ({self.name})"	×
UNCOV 384	self.wrappers = data.get("sequana_wrappers", "latest")	×
385	self.intro = intro	×
386
387	config.pipeline_version = data.get("pipeline_version", "latest")	×
388	config.pipeline_name = data.get("name", "undefined")	×
389	config.sequana_wrappers = self.wrappers	×
390
UNCOV 391	self.create_report_content(workflow=workflow)	×
392	self.create_html(output_filename)	×
393
394	def create_report_content(self, workflow=True):	3✔
395	"""Create the report content."""
396	self.sections = list()	×
397
UNCOV 398	for section in config.summary_sections:	×
UNCOV 399	self.sections.append(section)	×
400
401	if workflow:	×
UNCOV 402	self.workflow()	×
403	self.dependencies()	×
404	self.caller()	×
405
406	def workflow(self):	3✔
407	img = self.json["rulegraph"]	×
408	dag_svg = self.include_svg_image(img, alt="workflow")	×
409	snakefile = ".sequana/{}.rules".format(self.name)	×
410
UNCOV 411	try:	×
412	with open(snakefile, "r") as fp:	×
413	code = self.add_code_section(fp.read(), "python")	×
414	sf = self.create_hide_section("Sf", "Show/hide Snakemake file", code, hide=True)	×
UNCOV 415	sf = "\n".join(sf)	×
416	except IOError:	×
417	sf = "no snakefile found in .sequana/"	×
418
419	configfile = ".sequana/config.yaml"	×
420	try:	×
421	with open(configfile, "r") as fp:	×
422	code = self.add_code_section(fp.read(), "yaml")	×
UNCOV 423	c = self.create_hide_section("C", "Show/hide config file", code, hide=True)	×
424	c = "\n".join(c)	×
425	except IOError:	×
426	c = "no config found in .sequana/"	×
427
428	self.sections.append(	×
429	{
430	"name": "Workflow",
431	"anchor": "workflow",
432	"content": "<p>The following network shows the workflow of the pipeline. "
433	"Blue boxes are clickable and redirect to dedicated reports."
434	"</p>\n{0}\n"
435	"<p>The analysis was performed with the following "
436	'<a href="{3}">Snakemake</a> and <a href="{4}">configfile</a>:'
437	"</p>\n"
438	"<ul>\n"
439	" <li>{1}</li>\n"
440	" <li>{2}</li>\n"
441	"</ul>".format(dag_svg, sf, c, snakefile, configfile),
442	}
443	)
444
445	def get_table_versions(self):	3✔
446	"""Return third party tools from the requirements.txt and their versions."""
447
448	# if no version.txt is found, return nothing
UNCOV 449	if os.path.exists(".sequana/versions.txt") is False:	×
UNCOV 450	return ""	×
451
UNCOV 452	versions = []	×
UNCOV 453	tools = []	×
454	with open(".sequana/versions.txt", "r") as fin:	×
455	for line in fin.readlines():	×
UNCOV 456	tool, version = line.split()	×
457	versions.append(version)	×
458	tools.append(tool)	×
459
460	df = pd.DataFrame({"tool": tools, "version": versions})	×
461	try:	×
462	df["sort"] = df["tool"].str.lower()	×
463	df.sort_values(by="sort", axis=0, inplace=True)	×
UNCOV 464	df.drop("sort", axis=1, inplace=True)	×
465	except (KeyError, AttributeError): # could be empty	×
466	pass	×
467	datatable = DataTable(df, "dep_and_version")	×
468	datatable.datatable.datatable_options = {	×
469	"paging": "false",
470	"bFilter": "false",
471	"bInfo": "false",
472	"bSort": "false",
473	}
UNCOV 474	js = datatable.create_javascript_function()	×
UNCOV 475	html = datatable.create_datatable()	×
UNCOV 476	return js + "\n" + html	×
477
478	def dependencies(self):	3✔
479	"""Table with all python dependencies and a text file with tools
480	needed and their versions.
481	"""
482
UNCOV 483	html_table_versions = self.get_table_versions()	×
UNCOV 484	html_table_deps = self.get_table_dependencies()	×
UNCOV 485	pypi = self.create_link("Pypi", "http://pypi.python.org")	×
486
UNCOV 487	req = self.create_link("requirements", ".sequana/env.yml")	×
488
489	content = (	×
490	"<p>Third party tools can be found within containers (see config file abobe) if you use --use-apptainers option. Otherwise, here is a list of required dependencies and their versions.</p>"
491	"<p>{3}</p>"
492	"<p>Python dependencies (<b>{0}</b>){1}</p>".format(pypi, html_table_deps, req, html_table_versions)
493	)
494	l, c = self.create_hide_section("Dep", "collapse/expand", content, hide=True)	×
UNCOV 495	self.sections.append(	×
496	{
497	"name": "Dependencies {0}".format(self.add_float_right("<small>{0}</small>".format(l))),
498	"anchor": "dependencies",
499	"content": c,
500	}
501	)
502
503	def caller(self):	3✔
UNCOV 504	try:	×
UNCOV 505	with open(".sequana/info.txt") as fin:	×
UNCOV 506	command = fin.readlines()	×
UNCOV 507	command = "<pre>" + "\n".join([x for x in command if not x.startswith("#")]) + "</pre>"	×
UNCOV 508	except Exception as err:	×
UNCOV 509	print(err)	×
510	command = "unknown"	×
511	self.sections.append({"name": "Command", "anchor": "command", "content": command})	×

sequana / sequana / 16568275773

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