Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_quality_control():
report.files = {"sequana/quality_control":
[ { 'fn': sequana_data('summary_qc.json'), 'root': '.'}]
}
quality_control.MultiqcModule()
def clean_s_name(self, s_name, root):
""" Helper function to take a long file name and strip it
back to a clean sample name. Somewhat arbitrary.
:param s_name: The sample name to clean
:param root: The directory path that this file is within
:param prepend_dirs: boolean, whether to prepend dir name to s_name
:param trimmed: boolean, remove common trimming suffixes from name?
:return: The cleaned sample name, ready to be used
"""
if root is None:
root = ''
# Split then take first section to remove everything after these matches
for ext in config.fn_clean_exts:
s_name = s_name.split(ext ,1)[0]
if config.prepend_dirs:
s_name = "{} | {}".format(root.replace(os.sep, ' | '), s_name).lstrip('. | ')
return s_name
contigs_line.startswith("contigs:"),
bases_line.startswith("bases:"))):
return
# Get organism and sample name from the first line
# Assumes organism name only consists of two words,
# i.e. 'Genusname speciesname', and that the remaining
# text on the organism line is the sample name.
try:
organism = " ".join(first_line.strip().split(":", 1)[1].split()[:2])
s_name = self.clean_s_name(" ".join(first_line.split()[3:]), f['root'])
except KeyError:
organism = first_line.strip().split(":", 1)[1]
s_name = f['s_name']
# Don't try to guess sample name if requested in the config
if getattr(config, 'prokka_fn_snames', False):
s_name = f['s_name']
if s_name in self.prokka:
log.debug("Duplicate sample name found! Overwriting: {}".format(s_name))
self.prokka[s_name] = dict()
self.prokka[s_name]['organism'] = organism
self.prokka[s_name]['contigs'] = int(contigs_line.split(":")[1])
self.prokka[s_name]['bases'] = int(bases_line.split(":")[1])
# Get additional info from remaining lines
for line in f['f']:
description, value = line.split(":")
try:
self.prokka[s_name][description] = int(value)
except ValueError:
log.warning("Unable to parse line: '%s'", line)
config.skip_generalstats = True
config.exclude_modules = tuple(x for x in config.exclude_modules if x != 'general_stats')
run_modules = [m for m in run_modules if list(m.keys())[0] not in config.exclude_modules]
if len(run_modules) == 0:
logger.critical('No analysis modules specified!')
sys.exit(1)
run_module_names = [ list(m.keys())[0] for m in run_modules ]
logger.debug("Analysing modules: {}".format(', '.join(run_module_names)))
# Create the temporary working directories
tmp_dir = tempfile.mkdtemp()
logger.debug('Using temporary directory for creating report: {}'.format(tmp_dir))
config.data_tmp_dir = os.path.join(tmp_dir, 'multiqc_data')
if filename != 'stdout' and config.make_data_dir == True:
config.data_dir = config.data_tmp_dir
os.makedirs(config.data_dir)
else:
config.data_dir = None
config.plots_tmp_dir = os.path.join(tmp_dir, 'multiqc_plots')
if filename != 'stdout' and config.export_plots == True:
config.plots_dir = config.plots_tmp_dir
os.makedirs(config.plots_dir)
# Load the template
template_mod = config.avail_templates[config.template].load()
# Add an output subdirectory if specified by template
try:
config.output_dir = os.path.join(config.output_dir, template_mod.output_subdir)
except AttributeError:
pass # No subdirectory variable given
if not os.path.exists(os.path.dirname(config.output_fn)):
os.makedirs(os.path.dirname(config.output_fn))
logger.info("Report : {}".format(os.path.relpath(config.output_fn)))
if config.make_data_dir == False:
logger.info("Data : None")
else:
# Make directories for data_dir
logger.info("Data : {}".format(os.path.relpath(config.data_dir)))
if not os.path.exists(config.data_dir):
os.makedirs(config.data_dir)
# Modules have run, so data directory should be complete by now. Move its contents.
for f in os.listdir(config.data_tmp_dir):
fn = os.path.join(config.data_tmp_dir, f)
logger.debug("Moving data file from '{}' to '{}'".format(fn, config.data_dir))
shutil.move(fn, config.data_dir)
# Copy across the static plot images if requested
if config.export_plots:
config.plots_dir = os.path.join(config.output_dir, config.plots_dir_name)
if os.path.exists(config.plots_dir):
if config.force:
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.plots_dir)))
shutil.rmtree(config.plots_dir)
else:
logger.error("Output directory {} already exists.".format(config.plots_dir))
logger.info("Use -f or --force to overwrite existing reports")
shutil.rmtree(tmp_dir)
sys.exit(1)
os.makedirs(config.plots_dir)
logger.info("Plots : {}".format(os.path.relpath(config.plots_dir)))
run_modules = [m for m in run_modules if list(m.keys())[0] not in config.exclude_modules]
if len(run_modules) == 0:
logger.critical('No analysis modules specified!')
sys.exit(1)
run_module_names = [ list(m.keys())[0] for m in run_modules ]
logger.debug("Analysing modules: {}".format(', '.join(run_module_names)))
# Create the temporary working directories
tmp_dir = tempfile.mkdtemp()
logger.debug('Using temporary directory for creating report: {}'.format(tmp_dir))
config.data_tmp_dir = os.path.join(tmp_dir, 'multiqc_data')
if filename != 'stdout' and config.make_data_dir == True:
config.data_dir = config.data_tmp_dir
os.makedirs(config.data_dir)
else:
config.data_dir = None
config.plots_tmp_dir = os.path.join(tmp_dir, 'multiqc_plots')
if filename != 'stdout' and config.export_plots == True:
config.plots_dir = config.plots_tmp_dir
os.makedirs(config.plots_dir)
# Load the template
template_mod = config.avail_templates[config.template].load()
# Add an output subdirectory if specified by template
try:
config.output_dir = os.path.join(config.output_dir, template_mod.output_subdir)
except AttributeError:
pass # No subdirectory variable given
# Add custom content section names
# Make directories for report if needed
if not os.path.exists(os.path.dirname(config.output_fn)):
os.makedirs(os.path.dirname(config.output_fn))
logger.info("Report : {}".format(os.path.relpath(config.output_fn)))
if config.make_data_dir == False:
logger.info("Data : None")
else:
# Make directories for data_dir
logger.info("Data : {}".format(os.path.relpath(config.data_dir)))
if not os.path.exists(config.data_dir):
os.makedirs(config.data_dir)
# Modules have run, so data directory should be complete by now. Move its contents.
for f in os.listdir(config.data_tmp_dir):
fn = os.path.join(config.data_tmp_dir, f)
logger.debug("Moving data file from '{}' to '{}'".format(fn, config.data_dir))
shutil.move(fn, config.data_dir)
# Copy across the static plot images if requested
if config.export_plots:
config.plots_dir = os.path.join(config.output_dir, config.plots_dir_name)
if os.path.exists(config.plots_dir):
if config.force:
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.plots_dir)))
shutil.rmtree(config.plots_dir)
else:
logger.error("Output directory {} already exists.".format(config.plots_dir))
logger.info("Use -f or --force to overwrite existing reports")
shutil.rmtree(tmp_dir)
sys.exit(1)
os.makedirs(config.plots_dir)
logger.info("Plots : {}".format(os.path.relpath(config.plots_dir)))
if config.make_data_dir == False:
logger.info("Data : None")
else:
# Make directories for data_dir
logger.info("Data : {}".format(os.path.relpath(config.data_dir)))
if not os.path.exists(config.data_dir):
os.makedirs(config.data_dir)
# Modules have run, so data directory should be complete by now. Move its contents.
for f in os.listdir(config.data_tmp_dir):
fn = os.path.join(config.data_tmp_dir, f)
logger.debug("Moving data file from '{}' to '{}'".format(fn, config.data_dir))
shutil.move(fn, config.data_dir)
# Copy across the static plot images if requested
if config.export_plots:
config.plots_dir = os.path.join(config.output_dir, config.plots_dir_name)
if os.path.exists(config.plots_dir):
if config.force:
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.plots_dir)))
shutil.rmtree(config.plots_dir)
else:
logger.error("Output directory {} already exists.".format(config.plots_dir))
logger.info("Use -f or --force to overwrite existing reports")
shutil.rmtree(tmp_dir)
sys.exit(1)
os.makedirs(config.plots_dir)
logger.info("Plots : {}".format(os.path.relpath(config.plots_dir)))
# Modules have run, so plots directory should be complete by now. Move its contents.
for f in os.listdir(config.plots_tmp_dir):
fn = os.path.join(config.plots_tmp_dir, f)
logger.debug("Moving plots directory from '{}' to '{}'".format(fn, config.plots_dir))
shutil.move(fn, config.data_dir)
# Copy across the static plot images if requested
if config.export_plots:
config.plots_dir = os.path.join(config.output_dir, config.plots_dir_name)
if os.path.exists(config.plots_dir):
if config.force:
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.plots_dir)))
shutil.rmtree(config.plots_dir)
else:
logger.error("Output directory {} already exists.".format(config.plots_dir))
logger.info("Use -f or --force to overwrite existing reports")
shutil.rmtree(tmp_dir)
sys.exit(1)
os.makedirs(config.plots_dir)
logger.info("Plots : {}".format(os.path.relpath(config.plots_dir)))
# Modules have run, so plots directory should be complete by now. Move its contents.
for f in os.listdir(config.plots_tmp_dir):
fn = os.path.join(config.plots_tmp_dir, f)
logger.debug("Moving plots directory from '{}' to '{}'".format(fn, config.plots_dir))
shutil.move(fn, config.plots_dir)
plugin_hooks.mqc_trigger('before_template')
# Load in parent template files first if a child theme
try:
parent_template = config.avail_templates[template_mod.template_parent].load()
copy_tree(parent_template.template_dir, tmp_dir)
except AttributeError:
pass # Not a child theme
logger.info("Data : None")
else:
# Make directories for data_dir
logger.info("Data : {}".format(os.path.relpath(config.data_dir)))
if not os.path.exists(config.data_dir):
os.makedirs(config.data_dir)
# Modules have run, so data directory should be complete by now. Move its contents.
for f in os.listdir(config.data_tmp_dir):
fn = os.path.join(config.data_tmp_dir, f)
logger.debug("Moving data file from '{}' to '{}'".format(fn, config.data_dir))
shutil.move(fn, config.data_dir)
# Copy across the static plot images if requested
if config.export_plots:
config.plots_dir = os.path.join(config.output_dir, config.plots_dir_name)
if os.path.exists(config.plots_dir):
if config.force:
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.plots_dir)))
shutil.rmtree(config.plots_dir)
else:
logger.error("Output directory {} already exists.".format(config.plots_dir))
logger.info("Use -f or --force to overwrite existing reports")
shutil.rmtree(tmp_dir)
sys.exit(1)
os.makedirs(config.plots_dir)
logger.info("Plots : {}".format(os.path.relpath(config.plots_dir)))
# Modules have run, so plots directory should be complete by now. Move its contents.
for f in os.listdir(config.plots_tmp_dir):
fn = os.path.join(config.plots_tmp_dir, f)
logger.debug("Moving plots directory from '{}' to '{}'".format(fn, config.plots_dir))
shutil.move(fn, config.plots_dir)