#327

Committed 01 Apr 2025 09:36AM UTC coverage: 31.562% (+0.04%) from 31.526%

Build # #327

Build Type

Pull #204

travis-ci

Committed by

web-flow

Commit Message

Apply suggestions from code review

Pull Request Pull Request #204: microSALT - v4.2.0

Run Details

30 of 140 new or added lines in 7 files covered. (21.43%)

18 existing lines in 5 files now uncovered.

956 of 3029 relevant lines covered (31.56%)

0.32 hits per line

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

35.27

/microSALT/cli.py

"""This is the main entry point of microSALT.
   By: Isak Sylvin, @sylvinite"""

#!/usr/bin/env python

import click
import json
import os
import re
import subprocess
import sys
import yaml

from pkg_resources import iter_entry_points
from microSALT import __version__, preset_config, logger, wd
from microSALT.utils.scraper import Scraper
from microSALT.utils.job_creator import Job_Creator
from microSALT.utils.reporter import Reporter
from microSALT.utils.referencer import Referencer

default_sampleinfo = {
    "CG_ID_project": "XXX0000",
    "CG_ID_sample": "XXX0000A1",
    "Customer_ID_project": "100100",
    "Customer_ID_sample": "10XY123456",
    "Customer_ID": "cust000",
    "application_tag": "SOMTIN100",
    "date_arrival": "0001-01-01 00:00:00",
    "date_libprep": "0001-01-01 00:00:00",
    "date_sequencing": "0001-01-01 00:00:00",
    "method_libprep": "Not in LIMS",
    "method_sequencing": "Not in LIMS",
    "organism": "Staphylococcus aureus",
    "priority": "standard",
    "reference": "None",
}


if preset_config == "":
    click.echo(
        "ERROR - No properly set-up config under neither envvar MICROSALT_CONFIG nor ~/.microSALT/config.json. Exiting."
    )
    sys.exit(-1)


@click.pass_context
def set_cli_config(ctx, config):
    if config != "":
        if os.path.exists(config):
            try:
                t = ctx.obj["config"]
                with open(os.path.abspath(config), "r") as conf:
                    ctx.obj["config"] = json.load(conf)
                ctx.obj["config"]["folders"]["expec"] = t["folders"]["expec"]
                ctx.obj["config"]["folders"]["adapters"] = t["folders"]["adapters"]
                ctx.obj["config"]["config_path"] = os.path.abspath(config)
            except Exception as e:
                pass


def done():
    click.echo("INFO - Execution finished!")
    logger.debug("INFO - Execution finished!")


def review_sampleinfo(pfile):
    """Reviews sample info. Returns loaded json object"""

    try:
        with open(pfile) as json_file:
            data = json.load(json_file)
    except Exception as e:
        click.echo("Unable to read provided sample info file as json. Exiting..")
        sys.exit(-1)

    if isinstance(data, list):
        for entry in data:
            for k, v in default_sampleinfo.items():
                if not k in entry:
                    click.echo(
                        "WARNING - Parameter {} needs to be provided in sample json. Formatting example: ({})".format(
                            k, v
                        )
                    )
    else:
        for k, v in default_sampleinfo.items():
            if not k in data:
                click.echo(
                    "WARNING - Parameter {} needs to be provided in sample json. Formatting example: ({})".format(
                        k, v
                    )
                )
    return data


@click.group()
@click.version_option(__version__)
@click.pass_context
def root(ctx):
    """microbial Sequence Analysis and Loci-based Typing (microSALT) pipeline"""
    ctx.obj = {}
    ctx.obj["config"] = preset_config
    ctx.obj["log"] = logger


@root.command()
@click.argument("sampleinfo_file")
@click.option("--input", help="Full path to input folder", default="")
@click.option("--config", help="microSALT config to override default", default="")
@click.option(
    "--dry",
    help="Builds instance without posting to SLURM",
    default=False,
    is_flag=True,
)
@click.option(
    "--email",
    default=preset_config["regex"]["mail_recipient"],
    help="Forced e-mail recipient",
)
@click.option("--skip_update", default=False, help="Skips downloading of references", is_flag=True)
@click.option(
    "--force_update",
    default=False,
    help="Forces downloading of pubMLST references",
    is_flag=True,
)
@click.option("--untrimmed", help="Use untrimmed input data", default=False, is_flag=True)
@click.pass_context
def analyse(
    ctx,
    sampleinfo_file,
    input,
    config,
    dry,
    email,
    skip_update,
    force_update,
    untrimmed,
):
    """Sequence analysis, typing and resistance identification"""
    # Run section
    pool = []
    trimmed = not untrimmed
    set_cli_config(config)
    ctx.obj["config"]["regex"]["mail_recipient"] = email
    ctx.obj["config"]["dry"] = dry
    if not os.path.isdir(input):
        click.echo("ERROR - Sequence data folder {} does not exist.".format(input))
        ctx.abort()
    for subfolder in os.listdir(input):
        if os.path.isdir("{}/{}".format(input, subfolder)):
            pool.append(subfolder)

    run_settings = {
        "input": input,
        "dry": dry,
        "email": email,
        "skip_update": skip_update,
        "trimmed": not untrimmed,
        "pool": pool,
    }

    # Samples section
    sampleinfo = review_sampleinfo(sampleinfo_file)
    run_creator = Job_Creator(
        config=ctx.obj["config"],
        log=ctx.obj["log"],
        sampleinfo=sampleinfo,
        run_settings=run_settings,
    )

    ext_refs = Referencer(
        config=ctx.obj["config"],
        log=ctx.obj["log"],
        sampleinfo=sampleinfo,
        force=force_update,
    )
    click.echo("INFO - Checking versions of references..")
    try:
        if not skip_update:
            ext_refs.identify_new(project=True)
            ext_refs.update_refs()
            click.echo("INFO - Version check done. Creating sbatch jobs")
        else:
            click.echo("INFO - Skipping version check.")
    except Exception as e:
        click.echo("{}".format(e))
    if len(sampleinfo) > 1:
        run_creator.project_job()
    elif len(sampleinfo) == 1:
        run_creator.project_job(single_sample=True)
    else:
        ctx.abort()

    done()


@root.group()
@click.pass_context
def utils(ctx):
    """Utilities for specific purposes"""
    pass


@utils.group()
@click.pass_context
def refer(ctx):
    """Manipulates MLST organisms"""
    pass


@utils.command()
@click.argument("sampleinfo_file")
@click.option("--input", help="Full path to project folder", default="")
@click.option(
    "--track",
    help="Run a specific analysis track",
    default="default",
    type=click.Choice(["default", "typing", "qc", "cgmlst"]),
)
@click.option("--config", help="microSALT config to override default", default="")
@click.option(
    "--dry",
    help="Builds instance without posting to SLURM",
    default=False,
    is_flag=True,
)
@click.option(
    "--email",
    default=preset_config["regex"]["mail_recipient"],
    help="Forced e-mail recipient",
)
@click.option("--skip_update", default=False, help="Skips downloading of references", is_flag=True)
@click.option(
    "--report",
    default="default",
    type=click.Choice(["default", "typing", "motif_overview", "qc", "json_dump", "st_update"]),
)
@click.option("--output", help="Report output folder", default="")
@click.pass_context
def finish(ctx, sampleinfo_file, input, track, config, dry, email, skip_update, report, output):
    """Sequence analysis, typing and resistance identification"""
    # Run section
    pool = []
    set_cli_config(config)
    ctx.obj["config"]["regex"]["mail_recipient"] = email
    ctx.obj["config"]["dry"] = dry
    if not os.path.isdir(input):
        click.echo("ERROR - Sequence data folder {} does not exist.".format(input))
        ctx.abort()
    if output == "":
        output = input
    for subfolder in os.listdir(input):
        if os.path.isdir("{}/{}".format(input, subfolder)):
            pool.append(subfolder)

    run_settings = {
        "input": input,
        "track": track,
        "dry": dry,
        "email": email,
        "skip_update": skip_update,
    }

    # Samples section
    sampleinfo = review_sampleinfo(sampleinfo_file)
    ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], sampleinfo=sampleinfo)
    click.echo("INFO - Checking versions of references..")
    try:
        if not skip_update:
            ext_refs.identify_new(project=True)
            ext_refs.update_refs()
            click.echo("INFO - Version check done. Creating sbatch jobs")
        else:
            click.echo("INFO - Skipping version check.")
    except Exception as e:
        click.echo("{}".format(e))

    res_scraper = Scraper(
        config=ctx.obj["config"], log=ctx.obj["log"], sampleinfo=sampleinfo, input=input
    )
    if isinstance(sampleinfo, list) and len(sampleinfo) > 1:
        res_scraper.scrape_project()
        # for subfolder in pool:
        #  res_scraper.scrape_sample()
    else:
        res_scraper.scrape_sample()

    codemonkey = Reporter(
        config=ctx.obj["config"],
        log=ctx.obj["log"],
        sampleinfo=sampleinfo,
        output=output,
        collection=True,
    )
    codemonkey.report(report)
    done()


@refer.command()
@click.argument("organism")
@click.option("--force", help="Redownloads existing organism", default=False, is_flag=True)
@click.pass_context
def add(ctx, organism, force):
    """Adds a new internal organism from pubMLST"""
    referee = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], force=force)
    try:
        referee.add_pubmlst(organism)
    except Exception as e:
        click.echo(e.args[0])
        ctx.abort()
    click.echo("INFO - Checking versions of all references..")
    referee = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], force=force)
    referee.update_refs()


@refer.command()
@click.pass_context
def observe(ctx):
    """Lists all stored organisms"""
    refe = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])
    click.echo("INFO - Currently stored organisms:")
    for org in sorted(refe.existing_organisms()):
        click.echo(org.replace("_", " ").capitalize())


@utils.command()
@click.argument("sampleinfo_file")
@click.option(
    "--email",
    default=preset_config["regex"]["mail_recipient"],
    help="Forced e-mail recipient",
)
@click.option(
    "--type",
    default="default",
    type=click.Choice(["default", "typing", "motif_overview", "qc", "json_dump", "st_update"]),
)
@click.option("--output", help="Full path to output folder", default="")
@click.option("--collection", default=False, is_flag=True)
@click.pass_context
def report(ctx, sampleinfo_file, email, type, output, collection):
    """Re-generates report for a project"""
    ctx.obj["config"]["regex"]["mail_recipient"] = email
    sampleinfo = review_sampleinfo(sampleinfo_file)
    codemonkey = Reporter(
        config=ctx.obj["config"],
        log=ctx.obj["log"],
        sampleinfo=sampleinfo,
        output=output,
        collection=collection,
    )
    codemonkey.report(type)
    done()


@utils.command()
@click.pass_context
def view(ctx):
    """Starts an interactive webserver for viewing"""
    codemonkey = Reporter(config=ctx.obj["config"], log=ctx.obj["log"])
    codemonkey.start_web()


@utils.command()
@click.option("--input", help="Full path to project folder", default=os.getcwd())
@click.pass_context
def generate(ctx, input):
    """Creates a blank sample info json for the given input folder"""
    input = os.path.abspath(input)
    project_name = os.path.basename(input)

    defaults = default_sampleinfo.copy()

    pool = []
    if not os.path.isdir(input):
        click.echo("ERROR - Sequence data folder {} does not exist.".format(project_name))
        ctx.abort()
    elif input != os.getcwd():
        for subfolder in os.listdir(input):
            if os.path.isdir("{}/{}".format(input, subfolder)):
                pool.append(defaults.copy())
                pool[-1]["CG_ID_project"] = project_name
                pool[-1]["CG_ID_sample"] = subfolder
    else:
        project_name = "default_sample_info"
        pool.append(defaults.copy())

    with open("{}/{}.json".format(os.getcwd(), project_name), "w") as output:
        json.dump(pool, output, indent=2)
    click.echo("INFO - Created {}.json in current folder".format(project_name))
    done()


@utils.group()
@click.pass_context
def resync(ctx):
    """Updates internal ST with pubMLST equivalent"""


@resync.command()
@click.option(
    "--type",
    default="list",
    type=click.Choice(["report", "list"]),
    help="Output format",
)
@click.option("--customer", default="all", help="Customer id filter")
@click.option("--skip_update", default=False, help="Skips downloading of references", is_flag=True)
@click.option(
    "--email",
    default=preset_config["regex"]["mail_recipient"],
    help="Forced e-mail recipient",
)
@click.option("--output", help="Full path to output folder", default="")
@click.pass_context
def review(ctx, type, customer, skip_update, email, output):
    """Generates information about novel ST"""
    # Trace exists by some samples having pubMLST_ST filled in. Make trace function later
    ctx.obj["config"]["regex"]["mail_recipient"] = email
    ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])
    if not skip_update:
        ext_refs.update_refs()
        ext_refs.resync()
    click.echo("INFO - Version check done. Generating output")
    if type == "report":
        codemonkey = Reporter(config=ctx.obj["config"], log=ctx.obj["log"], output=output)
        codemonkey.report(type="st_update", customer=customer)
    elif type == "list":
        ext_refs.resync(type=type)
    done()


@resync.command()
@click.option("--force-update", default=False, is_flag=True, help="Forces update")
@click.pass_context
def update_refs(ctx, force_update: bool):
    """Updates all references"""
    ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], force=force_update)
    ext_refs.update_refs()
    done()


@resync.command()
@click.option("--force-update", default=False, is_flag=True, help="Forces update")
@click.pass_context
def update_from_static(ctx, force_update: bool):
    """Updates a specific organism"""
    ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])
    ext_refs.fetch_external(force=force_update)
    done()


@resync.command()
@click.argument("sample_name")
@click.option(
    "--force",
    default=False,
    is_flag=True,
    help="Resolves sample without checking for pubMLST match",
)
@click.pass_context
def overwrite(ctx, sample_name, force):
    """Flags sample as resolved"""
    ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])
    ext_refs.resync(type="overwrite", sample=sample_name, ignore=force)
    done()

1	"""This is the main entry point of microSALT.
2	By: Isak Sylvin, @sylvinite"""
3
4	#!/usr/bin/env python
5
6	import click	1✔
7	import json	1✔
8	import os	1✔
9	import re	1✔
10	import subprocess	1✔
11	import sys	1✔
12	import yaml	1✔
13
14	from pkg_resources import iter_entry_points	1✔
15	from microSALT import __version__, preset_config, logger, wd	1✔
16	from microSALT.utils.scraper import Scraper	1✔
17	from microSALT.utils.job_creator import Job_Creator	1✔
18	from microSALT.utils.reporter import Reporter	1✔
19	from microSALT.utils.referencer import Referencer	1✔
20
21	default_sampleinfo = {	1✔
22	"CG_ID_project": "XXX0000",
23	"CG_ID_sample": "XXX0000A1",
24	"Customer_ID_project": "100100",
25	"Customer_ID_sample": "10XY123456",
26	"Customer_ID": "cust000",
27	"application_tag": "SOMTIN100",
28	"date_arrival": "0001-01-01 00:00:00",
29	"date_libprep": "0001-01-01 00:00:00",
30	"date_sequencing": "0001-01-01 00:00:00",
31	"method_libprep": "Not in LIMS",
32	"method_sequencing": "Not in LIMS",
33	"organism": "Staphylococcus aureus",
34	"priority": "standard",
35	"reference": "None",
36	}
37
38
39	if preset_config == "":	1✔
40	click.echo(	×
41	"ERROR - No properly set-up config under neither envvar MICROSALT_CONFIG nor ~/.microSALT/config.json. Exiting."
42	)
43	sys.exit(-1)	×
44
45
46	@click.pass_context	1✔
47	def set_cli_config(ctx, config):
48	if config != "":	×
49	if os.path.exists(config):	×
50	try:	×
51	t = ctx.obj["config"]	×
52	with open(os.path.abspath(config), "r") as conf:	×
53	ctx.obj["config"] = json.load(conf)	×
54	ctx.obj["config"]["folders"]["expec"] = t["folders"]["expec"]	×
55	ctx.obj["config"]["folders"]["adapters"] = t["folders"]["adapters"]	×
56	ctx.obj["config"]["config_path"] = os.path.abspath(config)	×
57	except Exception as e:	×
58	pass	×
59
60
61	def done():	1✔
62	click.echo("INFO - Execution finished!")	×
63	logger.debug("INFO - Execution finished!")	×
64
65
66	def review_sampleinfo(pfile):	1✔
67	"""Reviews sample info. Returns loaded json object"""
68
69	try:	×
70	with open(pfile) as json_file:	×
71	data = json.load(json_file)	×
72	except Exception as e:	×
73	click.echo("Unable to read provided sample info file as json. Exiting..")	×
74	sys.exit(-1)	×
75
76	if isinstance(data, list):	×
77	for entry in data:	×
78	for k, v in default_sampleinfo.items():	×
79	if not k in entry:	×
80	click.echo(	×
81	"WARNING - Parameter {} needs to be provided in sample json. Formatting example: ({})".format(
82	k, v
83	)
84	)
85	else:
86	for k, v in default_sampleinfo.items():	×
87	if not k in data:	×
88	click.echo(	×
89	"WARNING - Parameter {} needs to be provided in sample json. Formatting example: ({})".format(
90	k, v
91	)
92	)
93	return data	×
94
95
96	@click.group()	1✔
97	@click.version_option(__version__)	1✔
98	@click.pass_context	1✔
99	def root(ctx):
100	"""microbial Sequence Analysis and Loci-based Typing (microSALT) pipeline"""
101	ctx.obj = {}	×
102	ctx.obj["config"] = preset_config	×
103	ctx.obj["log"] = logger	×
104
105
106	@root.command()	1✔
107	@click.argument("sampleinfo_file")	1✔
108	@click.option("--input", help="Full path to input folder", default="")	1✔
109	@click.option("--config", help="microSALT config to override default", default="")	1✔
110	@click.option(	1✔
111	"--dry",
112	help="Builds instance without posting to SLURM",
113	default=False,
114	is_flag=True,
115	)
116	@click.option(	1✔
117	"--email",
118	default=preset_config["regex"]["mail_recipient"],
119	help="Forced e-mail recipient",
120	)
121	@click.option("--skip_update", default=False, help="Skips downloading of references", is_flag=True)	1✔
122	@click.option(	1✔
123	"--force_update",
124	default=False,
125	help="Forces downloading of pubMLST references",
126	is_flag=True,
127	)
128	@click.option("--untrimmed", help="Use untrimmed input data", default=False, is_flag=True)	1✔
129	@click.pass_context	1✔
130	def analyse(
131	ctx,
132	sampleinfo_file,
133	input,
134	config,
135	dry,
136	email,
137	skip_update,
138	force_update,
139	untrimmed,
140	):
141	"""Sequence analysis, typing and resistance identification"""
142	# Run section
143	pool = []	×
144	trimmed = not untrimmed	×
145	set_cli_config(config)	×
146	ctx.obj["config"]["regex"]["mail_recipient"] = email	×
147	ctx.obj["config"]["dry"] = dry	×
148	if not os.path.isdir(input):	×
149	click.echo("ERROR - Sequence data folder {} does not exist.".format(input))	×
150	ctx.abort()	×
151	for subfolder in os.listdir(input):	×
152	if os.path.isdir("{}/{}".format(input, subfolder)):	×
153	pool.append(subfolder)	×
154
155	run_settings = {	×
156	"input": input,
157	"dry": dry,
158	"email": email,
159	"skip_update": skip_update,
160	"trimmed": not untrimmed,
161	"pool": pool,
162	}
163
164	# Samples section
165	sampleinfo = review_sampleinfo(sampleinfo_file)	×
166	run_creator = Job_Creator(	×
167	config=ctx.obj["config"],
168	log=ctx.obj["log"],
169	sampleinfo=sampleinfo,
170	run_settings=run_settings,
171	)
172
173	ext_refs = Referencer(	×
174	config=ctx.obj["config"],
175	log=ctx.obj["log"],
176	sampleinfo=sampleinfo,
177	force=force_update,
178	)
179	click.echo("INFO - Checking versions of references..")	×
180	try:	×
181	if not skip_update:	×
182	ext_refs.identify_new(project=True)	×
183	ext_refs.update_refs()	×
184	click.echo("INFO - Version check done. Creating sbatch jobs")	×
185	else:
186	click.echo("INFO - Skipping version check.")	×
187	except Exception as e:	×
188	click.echo("{}".format(e))	×
189	if len(sampleinfo) > 1:	×
190	run_creator.project_job()	×
191	elif len(sampleinfo) == 1:	×
192	run_creator.project_job(single_sample=True)	×
193	else:
194	ctx.abort()	×
195
196	done()	×
197
198
199	@root.group()	1✔
200	@click.pass_context	1✔
201	def utils(ctx):
202	"""Utilities for specific purposes"""
203	pass	×
204
205
206	@utils.group()	1✔
207	@click.pass_context	1✔
208	def refer(ctx):
209	"""Manipulates MLST organisms"""
210	pass	×
211
212
213	@utils.command()	1✔
214	@click.argument("sampleinfo_file")	1✔
215	@click.option("--input", help="Full path to project folder", default="")	1✔
216	@click.option(	1✔
217	"--track",
218	help="Run a specific analysis track",
219	default="default",
220	type=click.Choice(["default", "typing", "qc", "cgmlst"]),
221	)
222	@click.option("--config", help="microSALT config to override default", default="")	1✔
223	@click.option(	1✔
224	"--dry",
225	help="Builds instance without posting to SLURM",
226	default=False,
227	is_flag=True,
228	)
229	@click.option(	1✔
230	"--email",
231	default=preset_config["regex"]["mail_recipient"],
232	help="Forced e-mail recipient",
233	)
234	@click.option("--skip_update", default=False, help="Skips downloading of references", is_flag=True)	1✔
235	@click.option(	1✔
236	"--report",
237	default="default",
238	type=click.Choice(["default", "typing", "motif_overview", "qc", "json_dump", "st_update"]),
239	)
240	@click.option("--output", help="Report output folder", default="")	1✔
241	@click.pass_context	1✔
242	def finish(ctx, sampleinfo_file, input, track, config, dry, email, skip_update, report, output):
243	"""Sequence analysis, typing and resistance identification"""
244	# Run section
245	pool = []	×
246	set_cli_config(config)	×
247	ctx.obj["config"]["regex"]["mail_recipient"] = email	×
248	ctx.obj["config"]["dry"] = dry	×
249	if not os.path.isdir(input):	×
250	click.echo("ERROR - Sequence data folder {} does not exist.".format(input))	×
251	ctx.abort()	×
252	if output == "":	×
253	output = input	×
254	for subfolder in os.listdir(input):	×
255	if os.path.isdir("{}/{}".format(input, subfolder)):	×
256	pool.append(subfolder)	×
257
258	run_settings = {	×
259	"input": input,
260	"track": track,
261	"dry": dry,
262	"email": email,
263	"skip_update": skip_update,
264	}
265
266	# Samples section
267	sampleinfo = review_sampleinfo(sampleinfo_file)	×
NEW 268	ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], sampleinfo=sampleinfo)	×
UNCOV 269	click.echo("INFO - Checking versions of references..")	×
270	try:	×
271	if not skip_update:	×
272	ext_refs.identify_new(project=True)	×
273	ext_refs.update_refs()	×
274	click.echo("INFO - Version check done. Creating sbatch jobs")	×
275	else:
276	click.echo("INFO - Skipping version check.")	×
277	except Exception as e:	×
278	click.echo("{}".format(e))	×
279
280	res_scraper = Scraper(	×
281	config=ctx.obj["config"], log=ctx.obj["log"], sampleinfo=sampleinfo, input=input
282	)
283	if isinstance(sampleinfo, list) and len(sampleinfo) > 1:	×
284	res_scraper.scrape_project()	×
285	# for subfolder in pool:
286	# res_scraper.scrape_sample()
287	else:
288	res_scraper.scrape_sample()	×
289
290	codemonkey = Reporter(	×
291	config=ctx.obj["config"],
292	log=ctx.obj["log"],
293	sampleinfo=sampleinfo,
294	output=output,
295	collection=True,
296	)
297	codemonkey.report(report)	×
298	done()	×
299
300
301	@refer.command()	1✔
302	@click.argument("organism")	1✔
303	@click.option("--force", help="Redownloads existing organism", default=False, is_flag=True)	1✔
304	@click.pass_context	1✔
305	def add(ctx, organism, force):
306	"""Adds a new internal organism from pubMLST"""
307	referee = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], force=force)	×
308	try:	×
309	referee.add_pubmlst(organism)	×
310	except Exception as e:	×
311	click.echo(e.args[0])	×
312	ctx.abort()	×
313	click.echo("INFO - Checking versions of all references..")	×
314	referee = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], force=force)	×
315	referee.update_refs()	×
316
317
318	@refer.command()	1✔
319	@click.pass_context	1✔
320	def observe(ctx):
321	"""Lists all stored organisms"""
322	refe = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])	×
323	click.echo("INFO - Currently stored organisms:")	×
324	for org in sorted(refe.existing_organisms()):	×
325	click.echo(org.replace("_", " ").capitalize())	×
326
327
328	@utils.command()	1✔
329	@click.argument("sampleinfo_file")	1✔
330	@click.option(	1✔
331	"--email",
332	default=preset_config["regex"]["mail_recipient"],
333	help="Forced e-mail recipient",
334	)
335	@click.option(	1✔
336	"--type",
337	default="default",
338	type=click.Choice(["default", "typing", "motif_overview", "qc", "json_dump", "st_update"]),
339	)
340	@click.option("--output", help="Full path to output folder", default="")	1✔
341	@click.option("--collection", default=False, is_flag=True)	1✔
342	@click.pass_context	1✔
343	def report(ctx, sampleinfo_file, email, type, output, collection):
344	"""Re-generates report for a project"""
345	ctx.obj["config"]["regex"]["mail_recipient"] = email	×
346	sampleinfo = review_sampleinfo(sampleinfo_file)	×
347	codemonkey = Reporter(	×
348	config=ctx.obj["config"],
349	log=ctx.obj["log"],
350	sampleinfo=sampleinfo,
351	output=output,
352	collection=collection,
353	)
354	codemonkey.report(type)	×
355	done()	×
356
357
358	@utils.command()	1✔
359	@click.pass_context	1✔
360	def view(ctx):
361	"""Starts an interactive webserver for viewing"""
362	codemonkey = Reporter(config=ctx.obj["config"], log=ctx.obj["log"])	×
363	codemonkey.start_web()	×
364
365
366	@utils.command()	1✔
367	@click.option("--input", help="Full path to project folder", default=os.getcwd())	1✔
368	@click.pass_context	1✔
369	def generate(ctx, input):
370	"""Creates a blank sample info json for the given input folder"""
371	input = os.path.abspath(input)	×
372	project_name = os.path.basename(input)	×
373
374	defaults = default_sampleinfo.copy()	×
375
376	pool = []	×
377	if not os.path.isdir(input):	×
NEW 378	click.echo("ERROR - Sequence data folder {} does not exist.".format(project_name))	×
UNCOV 379	ctx.abort()	×
380	elif input != os.getcwd():	×
381	for subfolder in os.listdir(input):	×
382	if os.path.isdir("{}/{}".format(input, subfolder)):	×
383	pool.append(defaults.copy())	×
384	pool[-1]["CG_ID_project"] = project_name	×
385	pool[-1]["CG_ID_sample"] = subfolder	×
386	else:
387	project_name = "default_sample_info"	×
388	pool.append(defaults.copy())	×
389
390	with open("{}/{}.json".format(os.getcwd(), project_name), "w") as output:	×
391	json.dump(pool, output, indent=2)	×
392	click.echo("INFO - Created {}.json in current folder".format(project_name))	×
393	done()	×
394
395
396	@utils.group()	1✔
397	@click.pass_context	1✔
398	def resync(ctx):
399	"""Updates internal ST with pubMLST equivalent"""
400
401
402	@resync.command()	1✔
403	@click.option(	1✔
404	"--type",
405	default="list",
406	type=click.Choice(["report", "list"]),
407	help="Output format",
408	)
409	@click.option("--customer", default="all", help="Customer id filter")	1✔
410	@click.option("--skip_update", default=False, help="Skips downloading of references", is_flag=True)	1✔
411	@click.option(	1✔
412	"--email",
413	default=preset_config["regex"]["mail_recipient"],
414	help="Forced e-mail recipient",
415	)
416	@click.option("--output", help="Full path to output folder", default="")	1✔
417	@click.pass_context	1✔
418	def review(ctx, type, customer, skip_update, email, output):
419	"""Generates information about novel ST"""
420	# Trace exists by some samples having pubMLST_ST filled in. Make trace function later
421	ctx.obj["config"]["regex"]["mail_recipient"] = email	×
422	ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])	×
423	if not skip_update:	×
424	ext_refs.update_refs()	×
425	ext_refs.resync()	×
426	click.echo("INFO - Version check done. Generating output")	×
427	if type == "report":	×
NEW 428	codemonkey = Reporter(config=ctx.obj["config"], log=ctx.obj["log"], output=output)	×
UNCOV 429	codemonkey.report(type="st_update", customer=customer)	×
430	elif type == "list":	×
431	ext_refs.resync(type=type)	×
432	done()	×
433
434
435	@resync.command()	1✔
436	@click.option("--force-update", default=False, is_flag=True, help="Forces update")	1✔
437	@click.pass_context	1✔
438	def update_refs(ctx, force_update: bool):	1✔
439	"""Updates all references"""
NEW 440	ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"], force=force_update)	×
NEW 441	ext_refs.update_refs()	×
NEW 442	done()	×
443
444
445	@resync.command()	1✔
446	@click.option("--force-update", default=False, is_flag=True, help="Forces update")	1✔
447	@click.pass_context	1✔
448	def update_from_static(ctx, force_update: bool):	1✔
449	"""Updates a specific organism"""
NEW 450	ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])	×
NEW 451	ext_refs.fetch_external(force=force_update)	×
NEW 452	done()	×
453
454
455	@resync.command()	1✔
456	@click.argument("sample_name")	1✔
457	@click.option(	1✔
458	"--force",
459	default=False,
460	is_flag=True,
461	help="Resolves sample without checking for pubMLST match",
462	)
463	@click.pass_context	1✔
464	def overwrite(ctx, sample_name, force):
465	"""Flags sample as resolved"""
466	ext_refs = Referencer(config=ctx.obj["config"], log=ctx.obj["log"])	×
467	ext_refs.resync(type="overwrite", sample=sample_name, ignore=force)	×
468	done()	×

Clinical-Genomics / microSALT / #327

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