Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)))
# Copy over files if requested by the theme
try:
for f in template_mod.copy_files:
fn = os.path.join(tmp_dir, f)
dest_dir = os.path.join( os.path.dirname(config.output_fn), f)
copy_tree(fn, dest_dir)
except AttributeError:
pass # No files to copy
# Clean up temporary directory
shutil.rmtree(tmp_dir)
# Zip the data directory if requested
if config.zip_data_dir and config.data_dir is not None:
shutil.make_archive(config.data_dir, 'zip', config.data_dir)
shutil.rmtree(config.data_dir)
# Try to create a PDF if requested
if make_pdf:
try:
pdf_fn_name = config.output_fn.replace('.html', '.pdf')
pandoc_call = [
'pandoc',
'--standalone',
config.output_fn,
'--output', pdf_fn_name,
'--pdf-engine=xelatex',
'-V', 'documentclass=article',
'-V', 'geometry=margin=1in',
'-V', 'title='
]
plugin_hooks.mqc_trigger('before_report_generation')
# Data Export / MegaQC integration - save report data to file or send report data to an API endpoint
if (config.data_dump_file or config.megaqc_url) and config.megaqc_upload:
multiqc_json_dump = megaqc.multiqc_dump_json(report)
if config.data_dump_file:
util_functions.write_data_file(multiqc_json_dump, 'multiqc_data', False, 'json')
if config.megaqc_url:
megaqc.multiqc_api_post(multiqc_json_dump)
# Make the final report path & data directories
if filename != 'stdout':
config.output_fn = os.path.join(config.output_dir, config.output_fn_name)
config.data_dir = os.path.join(config.output_dir, config.data_dir_name)
# Check for existing reports and remove if -f was specified
if os.path.exists(config.output_fn) or (config.make_data_dir and os.path.exists(config.data_dir)):
if config.force:
if os.path.exists(config.output_fn):
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.output_fn)))
os.remove(config.output_fn)
if config.make_data_dir and os.path.exists(config.data_dir):
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.data_dir)))
shutil.rmtree(config.data_dir)
else:
# Set up the base names of the report and the data dir
report_num = 1
report_base, report_ext = os.path.splitext(config.output_fn_name)
dir_base = os.path.basename(config.data_dir)
# Iterate through appended numbers until we find one that's free
while os.path.exists(config.output_fn) or (config.make_data_dir and os.path.exists(config.data_dir)):
config.output_fn = os.path.join(config.output_dir, "{}_{}{}".format(report_base, report_num, report_ext) )
def write_csv_file(self, data, fn, sort_cols=False):
""" Write a tab-delimited data file to the reports directory.
:param: data - a 2D dict, first key sample name (row header),
second key field (column header).
:param: fn - Desired filename. Directory will be prepended automatically.
:return: None """
if config.data_dir is not None:
with io.open (os.path.join(config.data_dir, fn), "w", encoding='utf-8') as f:
print( report.dict_to_csv( data, sort_cols=sort_cols ), file=f)
config.output_fn_name = os.path.basename(config.output_fn)
config.data_dir_name = os.path.basename(config.data_dir)
logger.warning("Previous MultiQC output found! Adjusting filenames..")
logger.warning("Use -f or --force to overwrite existing reports instead")
# 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:
report.plot_compressed_json = report.compress_json(report.plot_data)
plugin_hooks.mqc_trigger('before_report_generation')
# Data Export / MegaQC integration - save report data to file or send report data to an API endpoint
if (config.data_dump_file or config.megaqc_url) and config.megaqc_upload:
multiqc_json_dump = megaqc.multiqc_dump_json(report)
if config.data_dump_file:
util_functions.write_data_file(multiqc_json_dump, 'multiqc_data', False, 'json')
if config.megaqc_url:
megaqc.multiqc_api_post(multiqc_json_dump)
# Make the final report path & data directories
if filename != 'stdout':
config.output_fn = os.path.join(config.output_dir, config.output_fn_name)
config.data_dir = os.path.join(config.output_dir, config.data_dir_name)
# Check for existing reports and remove if -f was specified
if os.path.exists(config.output_fn) or (config.make_data_dir and os.path.exists(config.data_dir)):
if config.force:
if os.path.exists(config.output_fn):
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.output_fn)))
os.remove(config.output_fn)
if config.make_data_dir and os.path.exists(config.data_dir):
logger.warning("Deleting : {} (-f was specified)".format(os.path.relpath(config.data_dir)))
shutil.rmtree(config.data_dir)
else:
# Set up the base names of the report and the data dir
report_num = 1
report_base, report_ext = os.path.splitext(config.output_fn_name)
dir_base = os.path.basename(config.data_dir)
# Iterate through appended numbers until we find one that's free
def write_csv_file(self, data, fn, sort_cols=False):
""" Write a tab-delimited data file to the reports directory.
:param: data - a 2D dict, first key sample name (row header),
second key field (column header).
:param: fn - Desired filename. Directory will be prepended automatically.
:return: None """
if config.data_dir is not None:
with io.open (os.path.join(config.data_dir, fn), "w", encoding='utf-8') as f:
print( report.dict_to_csv( data, sort_cols=sort_cols ), file=f)