12591050402

Committed 21 Nov 2024 10:05PM UTC coverage: 92.575%. Remained the same

Build # 12591050402

Build Type

push

github

Committed by

cokelaer

Commit Message

add --init-new-pipeline

Run Details

6 of 6 new or added lines in 1 file covered. (100.0%)

3 existing lines in 1 file now uncovered.

1359 of 1468 relevant lines covered (92.57%)

3.7 hits per line

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

97.73

/sequana_pipetools/scripts/main.py

#
#  This file is part of Sequana software
#
#  Copyright (c) 2016-2021 - Sequana Dev Team (https://sequana.readthedocs.io)
#
#  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
#  Contributors:  https://github.com/sequana/sequana/graphs/contributors
##############################################################################
import importlib
import os
import subprocess
import sys
import tempfile

import rich_click as click

from sequana_pipetools import version
from sequana_pipetools.misc import url2hash
from sequana_pipetools.snaketools.dot_parser import DOTParser
from sequana_pipetools.snaketools.errors import PipeError
from sequana_pipetools.snaketools.pipeline_utils import get_pipeline_statistics
from sequana_pipetools.snaketools.sequana_config import SequanaConfig

click.rich_click.USE_MARKDOWN = True
click.rich_click.SHOW_METAVARS_COLUMN = False
click.rich_click.APPEND_METAVARS_HELP = True
click.rich_click.STYLE_ERRORS_SUGGESTION = "magenta italic"
click.rich_click.SHOW_ARGUMENTS = True
click.rich_click.FOOTER_TEXT = (
    "Authors: Thomas Cokelaer, Dimitri Desvillechabrol -- http://github.com/sequana/sequana_pipetools"
)
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])


class ClickComplete:
    # For the -o default, this was an issue with compgen removing the slashes on
    # directrories . Solution was found here:
    # https://stackoverflow.com/questions/12933362/getting-compgen-to-include-slashes-on-directories-when-looking-for-files
    # Before using this option, a second directory could not be completed e.g.
    # in --databases, only the first argument could be completed, which was
    # really annoying.

    # KEEP '#version:' on first line since it is used in
    # sequana/pipeline_common.py right now
    setup = """#version: {version}
#info:
function _mycomplete_{pipeline_name}()
{{
    local cur prev opts
    COMPREPLY=()
    cur="${{COMP_WORDS[COMP_CWORD]}}"
    prev="${{COMP_WORDS[COMP_CWORD-1]}}"
    opts="{options}"
    case "${{prev}}" in
    """

    teardown = """
        #;;
    esac
    #if [[ ${{cur}} == -* ]] ; then
        COMPREPLY=( $(compgen -W "${{opts}}" -- ${{cur}}) )
        return 0
    #fi

}}
#complete -d -X '.[^./]*' -F _mycomplete_ sequana_{pipeline_name}
complete -o nospace -o default -F _mycomplete_{pipeline_name} sequana_{pipeline_name}
    """

    def __init__(self, pipeline_name):
        self._set_pipeline_name(pipeline_name)

    def _get_pipeline_name(self):
        return self._pipeline_name

    def _set_pipeline_name(self, name):
        self._pipeline_name = name
        self._init_config_file()
        self._init_version()

    pipeline_name = property(_get_pipeline_name, _set_pipeline_name)

    def save_completion_script(self):
        config_path = self.config_path
        pipeline_name = self.pipeline_name

        output_filename = f"{config_path}/{pipeline_name}.sh"

        arguments = self.get_arguments()

        with open(output_filename, "w") as fout:
            fout.write(
                self.setup.format(
                    version=self.pipeline_version, pipeline_name=pipeline_name, options=" ".join(arguments)
                )
            )
            for action in self._actions:
                name = action.opts[0]

                if action.type.name == "choice":
                    fout.write(self.set_option_with_choice(name, action.type.choices))
                elif action.type.name == "path":
                    fout.write(self.set_option_path(name))
                # elif action.type.name"-file"):
                #    fout.write(self.set_option_file(action.name))
                # elif action.name in ["--databases", "--from-project"]:
                #    fout.write(self.set_option_directory(action.name))
            fout.write(self.teardown.format(pipeline_name=self.pipeline_name))

    def _init_config_file(self):
        # do not import sequana, save time, let us use easydev
        from easydev import CustomConfig

        configuration = CustomConfig("sequana", verbose=False)
        sequana_config_path = configuration.user_config_dir
        path = sequana_config_path + os.sep + "pipelines"
        if os.path.exists(path):
            pass
        else:  # pragma: no cover
            os.mkdir(path)
        self.config_path = path
        return path

    def _init_version(self):
        importlib.import_module("sequana_pipelines.{}".format(self.pipeline_name))
        importlib.import_module("sequana_pipelines.{}.main".format(self.pipeline_name))
        module = sys.modules["sequana_pipelines.{}".format(self.pipeline_name)]
        version = module.version
        self.pipeline_version = version

    def get_arguments(self):
        importlib.import_module("sequana_pipelines.{}".format(self.pipeline_name))
        importlib.import_module("sequana_pipelines.{}.main".format(self.pipeline_name))
        module = sys.modules["sequana_pipelines.{}".format(self.pipeline_name)]

        main = module.__getattribute__("main")
        self._actions = [x for x in main.main.params]
        arguments = [f"{x.opts[0]}" for x in main.main.params]
        arguments = [x.replace("_", "-") for x in arguments]
        arguments = sorted(arguments)

        return arguments

    def set_option_with_choice(self, option_name, option_choices):

        option_choices = " ".join(option_choices)
        data = f"""
            {option_name})
                local choices="{option_choices}"
                COMPREPLY=( $(compgen -W "${{choices}}" -- ${{cur}}) )
                return 0
                ;;"""
        return data

    def set_option_path(self, option_name):
        data = f"""
            {option_name})
                xpat=".[!.]*"
                COMPREPLY=( $(compgen -X "${{xpat}}" -d ${{cur}}) )
                return 0
                ;;"""
        return data

    def set_option_file(self, option_name):
        data = f"""
            {option_name})
                COMPREPLY=( $(compgen  -f ${{cur}}) )
                return 0
                ;;"""
        return data


@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("--version", is_flag=True)
@click.option(
    "--dot2png", type=click.STRING, help="convert the input.dot into PNG file. Output name is called INPUT.sequana.png"
)
@click.option(
    "--completion",
    type=click.STRING,
    help="""Name of a pipelines for which you wish to create the completion file. Set to a valid Sequana pipeline name that must be installed""",
)
@click.option(
    "--force",
    is_flag=True,
    help="""overwrite files in sequana config pipeline directory (used with
--completion)""",
)
@click.option("--stats", is_flag=True, help="""Plot some stats related to the Sequana pipelines (installed)""")
@click.option(
    "--config-to-schema",
    type=click.Path(file_okay=True, dir_okay=False),
    help="""Given a config file, this command creates a draft schema file""",
)
@click.option("--slurm-diag", is_flag=True, help="Scans slurm files and get summary information")
@click.option("--url2hash", type=click.STRING, help="For developers. Convert a URL to hash mame. ")
@click.option(
    "--init-new-pipeline",
    is_flag=True,
    help="Give name of new pipeline and this will create full structure for a new sequana pipeline",
)
def main(**kwargs):
    """Pipetools utilities for the Sequana project (sequana.readthedocs.io)

    The pipetools package is dedicated to the developers of Sequana pipelines.
    However, Pipetools provides a set of utilities starting with the completion
    script for the different pipeline.

    To create a completion script, first install the pipeline::

        pip install sequana_multitax

    Then, type (--force to replace existing one):

        sequana_pipetools --completion multitax --force

    """

    if kwargs["version"]:
        click.echo(f"sequana_pipetools v{version}")
        return
    elif kwargs["url2hash"]:
        click.echo(url2hash(kwargs["url2hash"]))
    elif kwargs["dot2png"]:
        name = kwargs["dot2png"]
        assert name.endswith(".dot")
        outname = name.replace(".dot", ".sequana.png")
        with tempfile.NamedTemporaryFile(mode="w") as fout:
            d = DOTParser(name)
            d.add_urls(fout.name)
            cmd = f"dot -Tpng {fout.name} -o {outname}"
            subprocess.call(cmd.split())

    elif kwargs["completion"]:
        name = kwargs["completion"]

        if kwargs["force"] is True:
            choice = "y"
        else:  # pragma: no cover
            msg = f"This action will replace the {name}.sh file stored in ~/.config/sequana/pipelines. Do you want to proceed y/n: "
            choice = input(msg)
        if choice != "y":  # pragma: no cover
            sys.exit(0)

        try:
            c = ClickComplete(name)
            c.save_completion_script()
        except Exception:  # pragma: no cover
            click.echo(f"# Warning {name} could not be imported. Nothing done")
        finally:
            click.echo("Please source the files using:: \n")
            click.echo("    source ~/.config/sequana/pipelines/{}.sh".format(name))
            click.echo("\nto activate the completion. Add the line above in your environement")
    elif kwargs["stats"]:
        wrappers, rules = get_pipeline_statistics()
        click.echo("\n ==== Number of wrappers per pipeline")
        click.echo(wrappers.sum(axis=0))
        click.echo("\n ==== Number of time a wrapper is used")
        click.echo(wrappers.sum(axis=1))
        click.echo("\n ==== Number of rules used")
        click.echo(rules)
    elif kwargs["config_to_schema"]:
        config_file = kwargs["config_to_schema"]
        cfg = SequanaConfig(config_file)
        cfg.create_draft_schema()
    elif kwargs["slurm_diag"]:
        click.echo("Looking for slurm files")
        p = PipeError()
        p.status(".")
    elif kwargs["init_new_pipeline"]:  # pragma: no cover
        cmd = "cookiecutter https://github.com/sequana/sequana_pipeline_template -o . --overwrite-if-exists"
        subprocess.run(cmd.split(), capture_output=False)


if __name__ == "__main__":
    main()  # pragma: no cover

1	#
2	# This file is part of Sequana software
3	#
4	# Copyright (c) 2016-2021 - Sequana Dev Team (https://sequana.readthedocs.io)
5	#
6	# Distributed under the terms of the 3-clause BSD license.
7	# The full license is in the LICENSE file, distributed with this software.
8	#
9	# Website: https://github.com/sequana/sequana
10	# Documentation: http://sequana.readthedocs.io
11	# Contributors: https://github.com/sequana/sequana/graphs/contributors
12	##############################################################################
13	import importlib	4✔
14	import os	4✔
15	import subprocess	4✔
16	import sys	4✔
17	import tempfile	4✔
18
19	import rich_click as click	4✔
20
21	from sequana_pipetools import version	4✔
22	from sequana_pipetools.misc import url2hash	4✔
23	from sequana_pipetools.snaketools.dot_parser import DOTParser	4✔
24	from sequana_pipetools.snaketools.errors import PipeError	4✔
25	from sequana_pipetools.snaketools.pipeline_utils import get_pipeline_statistics	4✔
26	from sequana_pipetools.snaketools.sequana_config import SequanaConfig	4✔
27
28	click.rich_click.USE_MARKDOWN = True	4✔
29	click.rich_click.SHOW_METAVARS_COLUMN = False	4✔
30	click.rich_click.APPEND_METAVARS_HELP = True	4✔
31	click.rich_click.STYLE_ERRORS_SUGGESTION = "magenta italic"	4✔
32	click.rich_click.SHOW_ARGUMENTS = True	4✔
33	click.rich_click.FOOTER_TEXT = (	4✔
34	"Authors: Thomas Cokelaer, Dimitri Desvillechabrol -- http://github.com/sequana/sequana_pipetools"
35	)
36	CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])	4✔
37
38
39	class ClickComplete:	4✔
40	# For the -o default, this was an issue with compgen removing the slashes on
41	# directrories . Solution was found here:
42	# https://stackoverflow.com/questions/12933362/getting-compgen-to-include-slashes-on-directories-when-looking-for-files
43	# Before using this option, a second directory could not be completed e.g.
44	# in --databases, only the first argument could be completed, which was
45	# really annoying.
46
47	# KEEP '#version:' on first line since it is used in
48	# sequana/pipeline_common.py right now
49	setup = """#version: {version}	4✔
50	#info:
51	function _mycomplete_{pipeline_name}()
52	{{
53	local cur prev opts
54	COMPREPLY=()
55	cur="${{COMP_WORDS[COMP_CWORD]}}"
56	prev="${{COMP_WORDS[COMP_CWORD-1]}}"
57	opts="{options}"
58	case "${{prev}}" in
59	"""
60
61	teardown = """	4✔
62	#;;
63	esac
64	#if [[ ${{cur}} == -* ]] ; then
65	COMPREPLY=( $(compgen -W "${{opts}}" -- ${{cur}}) )
66	return 0
67	#fi
68
69	}}
70	#complete -d -X '.[^./]*' -F _mycomplete_ sequana_{pipeline_name}
71	complete -o nospace -o default -F _mycomplete_{pipeline_name} sequana_{pipeline_name}
72	"""
73
74	def __init__(self, pipeline_name):	4✔
75	self._set_pipeline_name(pipeline_name)	4✔
76
77	def _get_pipeline_name(self):	4✔
78	return self._pipeline_name	4✔
79
80	def _set_pipeline_name(self, name):	4✔
81	self._pipeline_name = name	4✔
82	self._init_config_file()	4✔
83	self._init_version()	4✔
84
85	pipeline_name = property(_get_pipeline_name, _set_pipeline_name)	4✔
86
87	def save_completion_script(self):	4✔
88	config_path = self.config_path	4✔
89	pipeline_name = self.pipeline_name	4✔
90
91	output_filename = f"{config_path}/{pipeline_name}.sh"	4✔
92
93	arguments = self.get_arguments()	4✔
94
95	with open(output_filename, "w") as fout:	4✔
96	fout.write(	4✔
97	self.setup.format(
98	version=self.pipeline_version, pipeline_name=pipeline_name, options=" ".join(arguments)
99	)
100	)
101	for action in self._actions:	4✔
102	name = action.opts[0]	4✔
103
104	if action.type.name == "choice":	4✔
105	fout.write(self.set_option_with_choice(name, action.type.choices))	4✔
106	elif action.type.name == "path":	4✔
107	fout.write(self.set_option_path(name))	4✔
108	# elif action.type.name"-file"):
109	# fout.write(self.set_option_file(action.name))
110	# elif action.name in ["--databases", "--from-project"]:
111	# fout.write(self.set_option_directory(action.name))
112	fout.write(self.teardown.format(pipeline_name=self.pipeline_name))	4✔
113
114	def _init_config_file(self):	4✔
115	# do not import sequana, save time, let us use easydev
116	from easydev import CustomConfig	4✔
117
118	configuration = CustomConfig("sequana", verbose=False)	4✔
119	sequana_config_path = configuration.user_config_dir	4✔
120	path = sequana_config_path + os.sep + "pipelines"	4✔
121	if os.path.exists(path):	4✔
UNCOV 122	pass	×
123	else: # pragma: no cover
124	os.mkdir(path)
125	self.config_path = path	4✔
126	return path	4✔
127
128	def _init_version(self):	4✔
129	importlib.import_module("sequana_pipelines.{}".format(self.pipeline_name))	4✔
130	importlib.import_module("sequana_pipelines.{}.main".format(self.pipeline_name))	4✔
131	module = sys.modules["sequana_pipelines.{}".format(self.pipeline_name)]	4✔
132	version = module.version	4✔
133	self.pipeline_version = version	4✔
134
135	def get_arguments(self):	4✔
136	importlib.import_module("sequana_pipelines.{}".format(self.pipeline_name))	4✔
137	importlib.import_module("sequana_pipelines.{}.main".format(self.pipeline_name))	4✔
138	module = sys.modules["sequana_pipelines.{}".format(self.pipeline_name)]	4✔
139
140	main = module.__getattribute__("main")	4✔
141	self._actions = [x for x in main.main.params]	4✔
142	arguments = [f"{x.opts[0]}" for x in main.main.params]	4✔
143	arguments = [x.replace("_", "-") for x in arguments]	4✔
144	arguments = sorted(arguments)	4✔
145
146	return arguments	4✔
147
148	def set_option_with_choice(self, option_name, option_choices):	4✔
149
150	option_choices = " ".join(option_choices)	4✔
151	data = f"""	4✔
152	{option_name})
153	local choices="{option_choices}"
154	COMPREPLY=( $(compgen -W "${{choices}}" -- ${{cur}}) )
155	return 0
156	;;"""
157	return data	4✔
158
159	def set_option_path(self, option_name):	4✔
160	data = f"""	4✔
161	{option_name})
162	xpat=".[!.]*"
163	COMPREPLY=( $(compgen -X "${{xpat}}" -d ${{cur}}) )
164	return 0
165	;;"""
166	return data	4✔
167
168	def set_option_file(self, option_name):	4✔
UNCOV 169	data = f"""	×
170	{option_name})
171	COMPREPLY=( $(compgen -f ${{cur}}) )
172	return 0
173	;;"""
UNCOV 174	return data	×
175
176
177	@click.command(context_settings=CONTEXT_SETTINGS)	4✔
178	@click.option("--version", is_flag=True)	4✔
179	@click.option(	4✔
180	"--dot2png", type=click.STRING, help="convert the input.dot into PNG file. Output name is called INPUT.sequana.png"
181	)
182	@click.option(	4✔
183	"--completion",
184	type=click.STRING,
185	help="""Name of a pipelines for which you wish to create the completion file. Set to a valid Sequana pipeline name that must be installed""",
186	)
187	@click.option(	4✔
188	"--force",
189	is_flag=True,
190	help="""overwrite files in sequana config pipeline directory (used with
191	--completion)""",
192	)
193	@click.option("--stats", is_flag=True, help="""Plot some stats related to the Sequana pipelines (installed)""")	4✔
194	@click.option(	4✔
195	"--config-to-schema",
196	type=click.Path(file_okay=True, dir_okay=False),
197	help="""Given a config file, this command creates a draft schema file""",
198	)
199	@click.option("--slurm-diag", is_flag=True, help="Scans slurm files and get summary information")	4✔
200	@click.option("--url2hash", type=click.STRING, help="For developers. Convert a URL to hash mame. ")	4✔
201	@click.option(	4✔
202	"--init-new-pipeline",
203	is_flag=True,
204	help="Give name of new pipeline and this will create full structure for a new sequana pipeline",
205	)
206	def main(**kwargs):	4✔
207	"""Pipetools utilities for the Sequana project (sequana.readthedocs.io)
208
209	The pipetools package is dedicated to the developers of Sequana pipelines.
210	However, Pipetools provides a set of utilities starting with the completion
211	script for the different pipeline.
212
213	To create a completion script, first install the pipeline::
214
215	pip install sequana_multitax
216
217	Then, type (--force to replace existing one):
218
219	sequana_pipetools --completion multitax --force
220
221	"""
222
223	if kwargs["version"]:	4✔
224	click.echo(f"sequana_pipetools v{version}")	4✔
225	return	4✔
226	elif kwargs["url2hash"]:	4✔
227	click.echo(url2hash(kwargs["url2hash"]))	4✔
228	elif kwargs["dot2png"]:	4✔
229	name = kwargs["dot2png"]	4✔
230	assert name.endswith(".dot")	4✔
231	outname = name.replace(".dot", ".sequana.png")	4✔
232	with tempfile.NamedTemporaryFile(mode="w") as fout:	4✔
233	d = DOTParser(name)	4✔
234	d.add_urls(fout.name)	4✔
235	cmd = f"dot -Tpng {fout.name} -o {outname}"	4✔
236	subprocess.call(cmd.split())	4✔
237
238	elif kwargs["completion"]:	4✔
239	name = kwargs["completion"]	4✔
240
241	if kwargs["force"] is True:	4✔
242	choice = "y"	4✔
243	else: # pragma: no cover
244	msg = f"This action will replace the {name}.sh file stored in ~/.config/sequana/pipelines. Do you want to proceed y/n: "
245	choice = input(msg)
246	if choice != "y": # pragma: no cover
247	sys.exit(0)
248
249	try:	4✔
250	c = ClickComplete(name)	4✔
251	c.save_completion_script()	4✔
252	except Exception: # pragma: no cover
253	click.echo(f"# Warning {name} could not be imported. Nothing done")
254	finally:
255	click.echo("Please source the files using:: \n")	4✔
256	click.echo(" source ~/.config/sequana/pipelines/{}.sh".format(name))	4✔
257	click.echo("\nto activate the completion. Add the line above in your environement")	4✔
258	elif kwargs["stats"]:	4✔
259	wrappers, rules = get_pipeline_statistics()	4✔
260	click.echo("\n ==== Number of wrappers per pipeline")	4✔
261	click.echo(wrappers.sum(axis=0))	4✔
262	click.echo("\n ==== Number of time a wrapper is used")	4✔
263	click.echo(wrappers.sum(axis=1))	4✔
264	click.echo("\n ==== Number of rules used")	4✔
265	click.echo(rules)	4✔
266	elif kwargs["config_to_schema"]:	4✔
267	config_file = kwargs["config_to_schema"]	4✔
268	cfg = SequanaConfig(config_file)	4✔
269	cfg.create_draft_schema()	4✔
270	elif kwargs["slurm_diag"]:	4✔
271	click.echo("Looking for slurm files")	4✔
272	p = PipeError()	4✔
273	p.status(".")	4✔
274	elif kwargs["init_new_pipeline"]: # pragma: no cover
275	cmd = "cookiecutter https://github.com/sequana/sequana_pipeline_template -o . --overwrite-if-exists"
276	subprocess.run(cmd.split(), capture_output=False)
277
278
279	if __name__ == "__main__":	4✔
280	main() # pragma: no cover

cokelaer / sequana_pipetools / 12591050402

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