Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_trigger_timeseries(workflow, singles, ifo_times, out_dir, special_tids=None,
exclude=None, require=None, tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
name = 'plot_trigger_timeseries'
secs = requirestr(workflow.cp.get_subsections(name), require)
secs = excludestr(secs, exclude)
files = FileList([])
for tag in secs:
node = PlotExecutable(workflow.cp, name, ifos=workflow.ifos,
out_dir=out_dir, tags=[tag] + tags).create_node()
node.add_multiifo_input_list_opt('--single-trigger-files', singles)
node.add_opt('--times', ifo_times)
node.new_output_file_opt(workflow.analysis_time, '.png', '--output-file')
if special_tids is not None:
node.add_opt('--special-trigger-ids', special_tids)
workflow += node
files += node.output_files
def make_singles_plot(workflow, trig_files, bank_file, veto_file, veto_name,
out_dir, exclude=None, require=None, tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
secs = requirestr(workflow.cp.get_subsections('plot_singles'), require)
secs = excludestr(secs, exclude)
files = FileList([])
for tag in secs:
for trig_file in trig_files:
node = PlotExecutable(workflow.cp, 'plot_singles',
ifos=trig_file.ifo,
out_dir=out_dir,
tags=[tag] + tags).create_node()
node.set_memory(15000)
node.add_input_opt('--bank-file', bank_file)
if veto_file is not None:
node.add_input_opt('--veto-file', veto_file)
node.add_opt('--segment-name', veto_name)
node.add_opt('--detector', trig_file.ifo)
def make_inference_samples_plot(
workflow, inference_file, output_dir, parameters=None,
name="inference_samples", analysis_seg=None, tags=None):
# default values
tags = [] if tags is None else tags
analysis_seg = workflow.analysis_time \
if analysis_seg is None else analysis_seg
# make the directory that will contain the output files
makedir(output_dir)
# make a node for plotting the posterior as a corner plot
node = PlotExecutable(workflow.cp, name, ifos=workflow.ifos,
out_dir=output_dir, universe="local",
tags=tags).create_node()
# add command line options
node.add_input_opt("--input-file", inference_file)
node.new_output_file_opt(analysis_seg, ".png", "--output-file")
node.add_opt("--parameters", " ".join(parameters))
# add node to workflow
workflow += node
return node.output_files
Tags to add to the inference executables.
Returns
-------
pycbc.workflow.FileList
A list of result and output files.
"""
# default values
tags = [] if tags is None else tags
analysis_seg = workflow.analysis_time \
if analysis_seg is None else analysis_seg
output_files = FileList([])
# make the directory that will contain the output files
makedir(output_dir)
# add command line options
for (ii, param) in enumerate(parameters):
plot_exe = PlotExecutable(workflow.cp, name, ifos=workflow.ifos,
out_dir=output_dir,
tags=tags+['param{}'.format(ii)])
node = plot_exe.create_node()
node.add_input_list_opt("--input-file", inference_files)
node.new_output_file_opt(analysis_seg, ".png", "--output-file")
node.add_opt("--parameters", param)
workflow += node
output_files += node.output_files
return output_files
def make_binned_hist(workflow, trig_file, veto_file, veto_name,
out_dir, bank_file, exclude=None,
require=None, tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
secs = requirestr(workflow.cp.get_subsections('plot_binnedhist'), require)
secs = excludestr(secs, exclude)
files = FileList([])
for tag in secs:
node = PlotExecutable(workflow.cp, 'plot_binnedhist',
ifos=trig_file.ifo,
out_dir=out_dir,
tags=[tag] + tags).create_node()
node.add_opt('--ifo', trig_file.ifo)
if veto_file is not None:
node.add_opt('--veto-segment-name', veto_name)
node.add_input_opt('--veto-file', veto_file)
node.add_input_opt('--trigger-file', trig_file)
node.add_input_opt('--bank-file', bank_file)
node.new_output_file_opt(trig_file.segment, '.png', '--output-file')
workflow += node
def make_gating_plot(workflow, insp_files, out_dir, tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
node = PlotExecutable(workflow.cp, 'plot_gating', ifos=workflow.ifos,
out_dir=out_dir, tags=tags).create_node()
node.add_input_list_opt('--input-file', insp_files)
node.new_output_file_opt(workflow.analysis_time, '.html', '--output-file')
workflow += node
def make_range_plot(workflow, psd_files, out_dir, exclude=None, require=None,
tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
secs = requirestr(workflow.cp.get_subsections('plot_range'), require)
secs = excludestr(secs, exclude)
files = FileList([])
for tag in secs:
node = PlotExecutable(workflow.cp, 'plot_range', ifos=workflow.ifos,
out_dir=out_dir, tags=[tag] + tags).create_node()
node.add_input_list_opt('--psd-files', psd_files)
node.new_output_file_opt(workflow.analysis_time, '.png', '--output-file')
workflow += node
files += node.output_files
return files
def make_throughput_plot(workflow, insp_files, out_dir, tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
node = PlotExecutable(workflow.cp, 'plot_throughput', ifos=workflow.ifos,
out_dir=out_dir, tags=tags).create_node()
node.add_input_list_opt('--input-file', insp_files)
node.new_output_file_opt(workflow.analysis_time, '.png', '--output-file')
workflow += node
def make_spectrum_plot(workflow, psd_files, out_dir, tags=None,
hdf_group=None, precalc_psd_files=None):
tags = [] if tags is None else tags
makedir(out_dir)
node = PlotExecutable(workflow.cp, 'plot_spectrum', ifos=workflow.ifos,
out_dir=out_dir, tags=tags).create_node()
node.add_input_list_opt('--psd-files', psd_files)
node.new_output_file_opt(workflow.analysis_time, '.png', '--output-file')
if hdf_group is not None:
node.add_opt('--hdf-group', hdf_group)
if precalc_psd_files is not None and len(precalc_psd_files) == 1:
node.add_input_list_opt('--psd-file', precalc_psd_files)
workflow += node
return node.output_files[0]
def make_coinc_snrchi_plot(workflow, inj_file, inj_trig, stat_file, trig_file,
out_dir, exclude=None, require=None, tags=None):
tags = [] if tags is None else tags
makedir(out_dir)
secs = requirestr(workflow.cp.get_subsections('plot_coinc_snrchi'), require)
secs = excludestr(secs, exclude)
files = FileList([])
for tag in secs:
node = PlotExecutable(workflow.cp, 'plot_coinc_snrchi', ifos=inj_trig.ifo,
out_dir=out_dir, tags=[tag] + tags).create_node()
node.add_input_opt('--found-injection-file', inj_file)
node.add_input_opt('--single-injection-file', inj_trig)
node.add_input_opt('--coinc-statistic-file', stat_file)
node.add_input_opt('--single-trigger-file', trig_file)
node.new_output_file_opt(inj_file.segment, '.png', '--output-file')
workflow += node
files += node.output_files
return files