Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • calibration/pycalibration
1 result
Show changes
Commits on Source (4)
Hide whitespace changes
Inline Side-by-side
cal_tools/cal_tools/tools.py
View file @ bc78a342
from collections import OrderedDict
import datetime
from glob import glob
from importlib.machinery import SourceFileLoader
import json
from os import chdir, environ, listdir, makedirs, path, remove, stat
from os.path import isdir, isfile, splitext
from os import environ, listdir, path, stat
from os.path import isfile, splitext
from queue import Queue
import re
from shutil import copy, copytree, move, rmtree
from subprocess import CalledProcessError, check_call, check_output
from textwrap import dedent
from time import sleep
from urllib.parse import urljoin
......@@ -20,318 +16,9 @@ from metadata_client.metadata_client import MetadataClient
from notebook.notebookapp import list_running_servers
import numpy as np
import requests
import tabulate
from jinja2 import Template
from .mdc_config import MDC_config
from .ana_tools import save_dict_to_hdf5
from xfel_calibrate import settings
def atoi(text):
"""
Convert string to integer is possible
:param text: string to be converted
:return: integer value or input string
"""
return int(text) if text.isdigit() else text
def natural_keys(text):
"""
Decompose string to list of integers and sub-strings
"""
return [atoi(c) for c in re.split(r'(\d+)', text)]
def combine_report(run_path, calibration):
sphinx_path = "{}/sphinx_rep".format(path.abspath(run_path))
makedirs(sphinx_path)
direntries = listdir(run_path)
direntries.sort(key=natural_keys)
for entry in direntries:
if isfile("{}/{}".format(run_path, entry)):
name, ext = splitext("{}".format(entry))
if ext == ".rst":
comps = name.split("__")
if len(comps) >= 3:
group, name_param, conc_param = "_".join(comps[:-2]), \
comps[-2], comps[-1]
else:
group, name_param, conc_param = comps[0], "None", "None"
with open("{}/{}.rst".format(sphinx_path, group),
"a") as gfile:
if conc_param != "None":
title = "{}, {} = {}".format(calibration, name_param,
conc_param)
gfile.write(title + "\n")
gfile.write("=" * len(title) + "\n")
gfile.write("\n")
with open("{}/{}".format(run_path, entry), "r") as ifile:
skip_next = False
for line in ifile.readlines():
if skip_next:
skip_next = False
continue
if conc_param == "None":
gfile.write(line)
elif " ".join(calibration.split()) != " ".join(
line.split()):
gfile.write(line)
else:
skip_next = True
gfile.write("\n\n")
if isdir("{}/{}".format(run_path, entry)):
copytree("{}/{}".format(run_path, entry),
"{}/{}".format(sphinx_path, entry))
return sphinx_path
def prepare_plots(run_path, threshold=1000000):
"""
Convert svg file to pdf or png to be used for latex
Conversion of svg to vector graphics pdf is performed using svglib3.
This procedure is CPU consuming. In order to speed up process
large svg files are converted to png format.
The links in the rst files are adapted accordingly to the
converted image files.
:param run_path: Run path of the slurm job
:param threshold: Max svg file size (in bytes) to be converted to pdf
"""
print('Convert svg to pdf and png')
run_path = path.abspath(run_path)
rst_files = glob('{}/*rst'.format(run_path))
for rst_file in rst_files:
rst_file_name = path.basename(rst_file)
rst_file_name = path.splitext(rst_file_name)[0]
svg_files = glob(
'{}/{}_files/*svg'.format(run_path, rst_file_name))
for f_path in svg_files:
f_name = path.basename(f_path)
f_name = path.splitext(f_name)[0]
if (stat(f_path)).st_size < threshold:
check_call(["svg2pdf", "{}".format(f_path)], shell=False)
new_ext = 'pdf'
else:
check_call(["convert", "{}".format(f_path),
"{}.png".format(f_name)], shell=False)
new_ext = 'png'
check_call(["sed",
"-i",
"s/{}.svg/{}.{}/g".format(f_name, f_name, new_ext),
rst_file],
shell=False)
def make_timing_summary(run_path, joblist):
"""
Create an rst file with timing summary of executed slurm jobs
:param run_path: Run path of the slurm job
:param joblist: List of slurm jobs
"""
print('Prepare timing summary')
run_path = path.abspath(run_path)
pars_vals = []
pars = 'JobID,Elapsed,Suspended'
pars_name = pars.split(',')
for job in joblist:
out = check_output(['sacct', '-j', job,
'--format={}'.format(pars)],
shell=False)
lines = str(out).split('\\n')
# loop over output lines, skip first two lines with header.
for line in lines[2:]:
s = line.split()
if len(s) == len(pars_name):
pars_vals.append(s)
break
tmpl = Template('''
Runtime summary
==============
.. math::
{% for line in table %}
{{ line }}
{%- endfor %}
''')
with open("{}/timing_summary.rst".format(run_path), "w+") as gfile:
if len(pars_vals) > 0:
table = tabulate.tabulate(pars_vals, tablefmt='latex',
headers=pars_name)
gfile.write(dedent(tmpl.render(table=table.split('\n'))))
def make_report(run_path, tmp_path, out_path, project, author, version,
report_to):
"""
Create calibration report (pdf file)
Automatically generated report document results, produced by
jupyter-notebooks.
:param run_path: Path to sphinx run directory
:param tmp_path: Run path of the slurm job
:param out_path: Output directory for report.
Overwritten if path to report is given in `report_to`
:param project: Project title
:param author: Author of the notebook
:param version: Version of the notebook
:param report_to: Name or path of the report file
"""
run_path = path.abspath(run_path)
report_path, report_name = path.split(report_to)
if report_path != '':
out_path = report_path
try:
check_call(["sphinx-quickstart",
"--quiet",
"--project='{}'".format(project),
"--author='{}'".format(author),
"-v", str(version),
"--suffix=.rst",
"--master=index",
"--ext-intersphinx",
"--ext-mathjax",
"--makefile",
"--no-batchfile", run_path])
except CalledProcessError:
raise Exception("Failed to run sphinx-quickbuild. Is sphinx installed?"
"Generated simple index.rst instead")
# quickbuild went well we need to edit the index.rst and conf.py files
module_path = "{}".format(path.abspath(path.dirname(__file__)))
conf = SourceFileLoader("conf",
"{}/conf.py".format(run_path)).load_module()
l_var = [v for v in dir(conf) if not v.startswith('__')]
with open("{}/conf.py.tmp".format(run_path), "w") as mf:
latex_elements = {'extraclassoptions': ',openany, oneside',
'preamble': r'\usepackage{longtable}',
'maketitle': r'\input{titlepage.tex.txt}'}
mf.write("latex_elements = {}\n".format(latex_elements))
mf.write("latex_logo = '{}/{}'\n".format(module_path,
settings.logo_path))
mf.write("latex_additional_files = ['titlepage.tex.txt']\n")
for var in l_var:
if var in ['latex_elements', 'latex_logo',
'latex_additional_files']:
continue
tmpl = '{} = {}\n'
v = getattr(conf, var, None)
if isinstance(v, str):
tmpl = '{} = "{}"\n'
# Set name of the latex document
if var == 'latex_documents' and len(v[0]) > 1:
v[0] = v[0][:1] + ('{}.tex'.format(report_name), ) + v[0][2:]
mf.write(tmpl.format(var, v))
remove("{}/conf.py".format(run_path))
move("{}/conf.py.tmp".format(run_path), "{}/conf.py".format(run_path))
direntries = listdir(run_path)
files_to_handle = []
for entry in direntries:
if isfile("{}/{}".format(run_path, entry)):
name, ext = splitext("{}".format(entry))
if ext == ".rst" and "index" not in name:
files_to_handle.append(name.strip())
index_tmp = Template('''
Calibration report
==================
.. toctree::
:maxdepth: 2
{% for k in keys %}
{{ k }}
{%- endfor %}
''')
with open("{}/index.rst".format(run_path), "w+") as mf:
mf.write(dedent(index_tmp.render(keys=files_to_handle)))
# finally call the make scripts
chdir(run_path)
try:
check_call(["make", "latexpdf"])
except CalledProcessError:
print("Failed to make pdf documentation")
print("Temp files will not be deleted and " +
"can be inspected at: {}".format(run_path))
return
print("Moving report to final location: {}".format(out_path))
copy('{}/_build/latex/{}.pdf'.format(run_path, report_name), out_path)
print("Removing temporary files at: {}".format(tmp_path))
rmtree(tmp_path)
def make_titlepage(sphinx_path, project, data_path, version):
"""
Create title page for report using template
:param sphinx_path: path to sphinx run directory
:param project: title of the project
:param data_path: path to input data sample used for notebook
:param version: Version of the pycalibration tool
"""
module_path = "{}".format(path.abspath(path.dirname(__file__)))
with open('{}/titlepage.tmpl'.format(module_path)) as file_:
title_tmp = Template(file_.read())
with open("{}/titlepage.tex.txt".format(sphinx_path), "w+") as mf:
mf.write(dedent(title_tmp.render(project=tex_escape(project),
data_path=tex_escape(data_path),
version=tex_escape(version))))
def finalize(joblist, finaljob, run_path, out_path, project, calibration,
author, version, report_to, data_path='Unknown'):
print("Waiting on jobs to finish: {}".format(joblist))
while True:
found_jobs = set()
output = check_output(['squeue']).decode('utf8')
for line in output.split("\n"):
for job in joblist:
if str(job) in line:
found_jobs.add(job)
if len(found_jobs) == 0:
break
sleep(10)
prepare_plots(run_path)
make_timing_summary(run_path, joblist + [str(finaljob)])
sphinx_path = combine_report(run_path, calibration)
make_titlepage(sphinx_path, project, data_path, version)
make_report(sphinx_path, run_path, out_path, project, author, version,
report_to)
def parse_runs(runs, return_type=str):
pruns = []
......@@ -716,30 +403,3 @@ def get_constant_from_db_and_time(device, constant, condition, empty_constant,
return data, m.calibration_constant_version.begin_at
else:
return data, None
def tex_escape(text):
"""
Escape latex special characters found in the text
:param text: a plain text message
:return: the message escaped to appear correctly in LaTeX
"""
conv = {
'&': r'\&',
'%': r'\%',
'$': r'\$',
'#': r'\#',
'_': r'\_',
'{': r'\{',
'}': r'\}',
'~': r'\textasciitilde{}',
'^': r'\^{}',
'\\': r'\textbackslash{}',
'<': r'\textless{}',
'>': r'\textgreater{}',
}
key_list = sorted(conv.keys(), key=lambda item: - len(item))
regex = re.compile('|'.join(re.escape(str(key)) for key in key_list))
return regex.sub(lambda match: conv[match.group()], text)
cal_tools/setup.py
View file @ bc78a342
......@@ -4,7 +4,6 @@ setup(
name='calTools',
version='0.1',
packages=['cal_tools',],
package_data={'': ['titlepage.tmpl', 'xfel.pdf']},
url='',
license='(c) European XFEL GmbH',
author='Steffen Hauf',
......
setup.py
View file @ bc78a342
......@@ -52,7 +52,8 @@ setup(
'xfel_calibrate.notebooks': 'xfel_calibrate/notebooks',
},
package_data={
'xfel_calibrate': ['bin/*.sh']+data_files
'xfel_calibrate': ['bin/*.sh'] + data_files + ['titlepage.tmpl',
'xfel.pdf']
},
cmdclass={
......
xfel_calibrate/calibrate.py
View file @ bc78a342
......@@ -19,9 +19,10 @@ import warnings
from .settings import *
from .notebooks import notebooks
from jinja2 import Template
import stat
import textwrap
from cal_tools.tools import tex_escape
from .finalize import tex_escape
# Add a class combining raw description formatting with
......@@ -491,6 +492,36 @@ def set_figure_format(nb, enable_vector_format):
cell.source += "\n%config InlineBackend.figure_formats = ['svg']\n"
def create_finalize_script(fmt_args, temp_path, job_list):
"""
Create a finalize script to produce output report
:param fmt_args: Dictionary of fmt arguments
:param temp_path: Path to temopary folder to run slurm job
:param job_list: List of slurm jobs
"""
tmpl = Template('''
#!/bin/tcsh
source /etc/profile.d/modules.sh
module load texlive
echo 'Running finalize script'
python3 -c "from xfel_calibrate.finalize import finalize;
finalize({{joblist}}, $1, '{{run_path}}', '{{out_path}}',
'{{project}}', '{{calibration}}', '{{author}}',
'{{version}}', '{{report_to}}', '{{in_folder}}' )"
''')
fmt_args['joblist'] = job_list
f_name = "{}/finalize.sh".format(temp_path)
with open(f_name, "w") as finfile:
finfile.write(textwrap.dedent(tmpl.render(**fmt_args)))
# change rights of the file to be:
# executed and writable for user, readable for user, group and others
all_stats = stat.S_IXUSR | stat.S_IWUSR | stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
os.chmod(f_name, all_stats)
def get_launcher_command(args, temp_path, dependent, job_list):
"""
Return a slurm launcher command
......@@ -529,16 +560,16 @@ def get_launcher_command(args, temp_path, dependent, job_list):
launcher_slurm += " --mem {}G".format(args.get('slurm_mem', '500'))
if dependent:
srun_dep = "--dependency=afterok"
srun_dep = " --dependency=afterok"
for jobid in job_list:
srun_dep += ":{}".format(jobid)
launcher_slurm += [srun_dep]
launcher_slurm += srun_dep
return launcher_slurm.split()
def concurrent_run(temp_path, nb, nbname, args, cparm=None, cval=None,
final_job=False, job_list=[], fmtcmd="", cluster_cores=8,
final_job=False, job_list=[], fmt_args={}, cluster_cores=8,
sequential=False, dependent=False,
show_title=True):
""" Launch a concurrent job on the cluster via SLURM
......@@ -570,16 +601,7 @@ def concurrent_run(temp_path, nb, nbname, args, cparm=None, cval=None,
# add finalization to the last job
if final_job:
import stat
with open("{}/finalize.sh".format(temp_path), "w") as finfile:
finfile.write("#!/bin/tcsh\n")
finfile.write("source /etc/profile.d/modules.sh\n")
finfile.write("module load texlive\n")
finfile.write("echo 'Running finalize script'\n")
finfile.write(
"python3 -c {}\n".format(fmtcmd.format(joblist=job_list)))
all_stats = stat.S_IXUSR | stat.S_IWUSR | stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH
os.chmod("{}/finalize.sh".format(temp_path), all_stats)
create_finalize_script(fmt_args, temp_path, job_list)
# then run an sbatch job
srun_base = []
......@@ -775,10 +797,6 @@ def run():
out_path))
os.makedirs(out_path, exist_ok=True)
cmd = ('"from cal_tools.tools import finalize; ' +
'finalize({{joblist}}, $1, \'{run_path}\', \'{out_path}\', ' +
'\'{project}\', \'{calibration}\', \'{author}\', '
'\'{version}\', \'{report_to}\', \'{in_folder}\' )"')
report_to = title.replace(" ", "")
if args["report_to"] is not None:
......@@ -786,10 +804,14 @@ def run():
folder = get_par_attr(parms, 'in_folder', 'value', '')
fmtcmd = cmd.format(run_path=run_tmp_path, out_path=out_path,
project=title, calibration=title,
author=author, version=version,
report_to=report_to, in_folder=folder)
fmt_args = {'run_path': run_tmp_path,
'out_path': out_path,
'project': title,
'calibration': title,
'author': author,
'version': version,
'report_to': report_to,
'in_folder': folder}
joblist = []
if concurrency.get("parameter", None) is None:
......@@ -798,7 +820,7 @@ def run():
jobid = concurrent_run(run_tmp_path, nb,
os.path.basename(notebook), args,
final_job=True, job_list=joblist,
fmtcmd=fmtcmd,
fmt_args=fmt_args,
cluster_cores=cluster_cores,
sequential=sequential)
......@@ -860,7 +882,7 @@ def run():
jobid = concurrent_run(run_tmp_path, nb, notebook, args,
cvar, cval, final_job,
joblist, fmtcmd,
joblist, fmt_args,
cluster_cores=cluster_cores,
sequential=sequential,
show_title=show_title)
......@@ -877,7 +899,7 @@ def run():
os.path.basename(notebook),
args,
final_job=final_job,
job_list=joblist, fmtcmd=fmtcmd,
job_list=joblist, fmt_args=fmt_args,
cluster_cores=cluster_cores,
sequential=sequential,
dependent=True)
......
xfel_calibrate/finalize.py 0 → 100644
View file @ bc78a342
from glob import glob
from importlib.machinery import SourceFileLoader
from os import chdir, listdir, makedirs, path, remove, stat
from os.path import isdir, isfile, splitext
import re
from shutil import copy, copytree, move, rmtree
from subprocess import CalledProcessError, check_call, check_output
from textwrap import dedent
from time import sleep
import tabulate
from jinja2 import Template
from .settings import *
def atoi(text):
"""
Convert string to integer is possible
:param text: string to be converted
:return: integer value or input string
"""
return int(text) if text.isdigit() else text
def natural_keys(text):
"""
Decompose string to list of integers and sub-strings
"""
return [atoi(c) for c in re.split(r'(\d+)', text)]
def combine_report(run_path, calibration):
sphinx_path = "{}/sphinx_rep".format(path.abspath(run_path))
makedirs(sphinx_path)
direntries = listdir(run_path)
direntries.sort(key=natural_keys)
for entry in direntries:
if isfile("{}/{}".format(run_path, entry)):
name, ext = splitext("{}".format(entry))
if ext == ".rst":
comps = name.split("__")
if len(comps) >= 3:
group, name_param, conc_param = "_".join(comps[:-2]), \
comps[-2], comps[-1]
else:
group, name_param, conc_param = comps[0], "None", "None"
with open("{}/{}.rst".format(sphinx_path, group),
"a") as gfile:
if conc_param != "None":
title = "{}, {} = {}".format(calibration, name_param,
conc_param)
gfile.write(title + "\n")
gfile.write("=" * len(title) + "\n")
gfile.write("\n")
with open("{}/{}".format(run_path, entry), "r") as ifile:
skip_next = False
for line in ifile.readlines():
if skip_next:
skip_next = False
continue
if conc_param == "None":
gfile.write(line)
elif " ".join(calibration.split()) != " ".join(
line.split()):
gfile.write(line)
else:
skip_next = True
gfile.write("\n\n")
if isdir("{}/{}".format(run_path, entry)):
copytree("{}/{}".format(run_path, entry),
"{}/{}".format(sphinx_path, entry))
return sphinx_path
def prepare_plots(run_path, threshold=1000000):
"""
Convert svg file to pdf or png to be used for latex
Conversion of svg to vector graphics pdf is performed using svglib3.
This procedure is CPU consuming. In order to speed up process
large svg files are converted to png format.
The links in the rst files are adapted accordingly to the
converted image files.
:param run_path: Run path of the slurm job
:param threshold: Max svg file size (in bytes) to be converted to pdf
"""
print('Convert svg to pdf and png')
run_path = path.abspath(run_path)
rst_files = glob('{}/*rst'.format(run_path))
for rst_file in rst_files:
rst_file_name = path.basename(rst_file)
rst_file_name = path.splitext(rst_file_name)[0]
svg_files = glob(
'{}/{}_files/*svg'.format(run_path, rst_file_name))
for f_path in svg_files:
f_name = path.basename(f_path)
f_name = path.splitext(f_name)[0]
if (stat(f_path)).st_size < threshold:
check_call(["svg2pdf", "{}".format(f_path)], shell=False)
new_ext = 'pdf'
else:
check_call(["convert", "{}".format(f_path),
"{}.png".format(f_name)], shell=False)
new_ext = 'png'
check_call(["sed",
"-i",
"s/{}.svg/{}.{}/g".format(f_name, f_name, new_ext),
rst_file],
shell=False)
def make_timing_summary(run_path, joblist):
"""
Create an rst file with timing summary of executed slurm jobs
:param run_path: Run path of the slurm job
:param joblist: List of slurm jobs
"""
print('Prepare timing summary')
run_path = path.abspath(run_path)
pars_vals = []
pars = 'JobID,Elapsed,Suspended'
pars_name = pars.split(',')
for job in joblist:
out = check_output(['sacct', '-j', job,
'--format={}'.format(pars)],
shell=False)
lines = str(out).split('\\n')
# loop over output lines, skip first two lines with header.
for line in lines[2:]:
s = line.split()
if len(s) == len(pars_name):
pars_vals.append(s)
break
tmpl = Template('''
Runtime summary
==============
.. math::
{% for line in table %}
{{ line }}
{%- endfor %}
''')
with open("{}/timing_summary.rst".format(run_path), "w+") as gfile:
if len(pars_vals) > 0:
table = tabulate.tabulate(pars_vals, tablefmt='latex',
headers=pars_name)
gfile.write(dedent(tmpl.render(table=table.split('\n'))))
def make_report(run_path, tmp_path, out_path, project, author, version,
report_to):
"""
Create calibration report (pdf file)
Automatically generated report document results, produced by
jupyter-notebooks.
:param run_path: Path to sphinx run directory
:param tmp_path: Run path of the slurm job
:param out_path: Output directory for report.
Overwritten if path to report is given in `report_to`
:param project: Project title
:param author: Author of the notebook
:param version: Version of the notebook
:param report_to: Name or path of the report file
"""
run_path = path.abspath(run_path)
report_path, report_name = path.split(report_to)
if report_path != '':
out_path = report_path
try:
check_call(["sphinx-quickstart",
"--quiet",
"--project='{}'".format(project),
"--author='{}'".format(author),
"-v", str(version),
"--suffix=.rst",
"--master=index",
"--ext-intersphinx",
"--ext-mathjax",
"--makefile",
"--no-batchfile", run_path])
except CalledProcessError:
raise Exception("Failed to run sphinx-quickbuild. Is sphinx installed?"
"Generated simple index.rst instead")
# quickbuild went well we need to edit the index.rst and conf.py files
module_path = "{}".format(path.abspath(path.dirname(__file__)))
conf = SourceFileLoader("conf",
"{}/conf.py".format(run_path)).load_module()
l_var = [v for v in dir(conf) if not v.startswith('__')]
with open("{}/conf.py.tmp".format(run_path), "w") as mf:
latex_elements = {'extraclassoptions': ',openany, oneside',
'preamble': r'\usepackage{longtable}',
'maketitle': r'\input{titlepage.tex.txt}'}
mf.write("latex_elements = {}\n".format(latex_elements))
mf.write("latex_logo = '{}/{}'\n".format(module_path,
logo_path))
mf.write("latex_additional_files = ['titlepage.tex.txt']\n")
for var in l_var:
if var in ['latex_elements', 'latex_logo',
'latex_additional_files']:
continue
tmpl = '{} = {}\n'
v = getattr(conf, var, None)
if isinstance(v, str):
tmpl = '{} = "{}"\n'
# Set name of the latex document
if var == 'latex_documents' and len(v[0]) > 1:
v[0] = v[0][:1] + ('{}.tex'.format(report_name), ) + v[0][2:]
mf.write(tmpl.format(var, v))
remove("{}/conf.py".format(run_path))
move("{}/conf.py.tmp".format(run_path), "{}/conf.py".format(run_path))
direntries = listdir(run_path)
files_to_handle = []
for entry in direntries:
if isfile("{}/{}".format(run_path, entry)):
name, ext = splitext("{}".format(entry))
if ext == ".rst" and "index" not in name:
files_to_handle.append(name.strip())
index_tmp = Template('''
Calibration report
==================
.. toctree::
:maxdepth: 2
{% for k in keys %}
{{ k }}
{%- endfor %}
''')
with open("{}/index.rst".format(run_path), "w+") as mf:
mf.write(dedent(index_tmp.render(keys=files_to_handle)))
# finally call the make scripts
chdir(run_path)
try:
check_call(["make", "latexpdf"])
except CalledProcessError:
print("Failed to make pdf documentation")
print("Temp files will not be deleted and " +
"can be inspected at: {}".format(run_path))
return
print("Moving report to final location: {}".format(out_path))
copy('{}/_build/latex/{}.pdf'.format(run_path, report_name), out_path)
print("Removing temporary files at: {}".format(tmp_path))
rmtree(tmp_path)
def make_titlepage(sphinx_path, project, data_path, version):
"""
Create title page for report using template
:param sphinx_path: path to sphinx run directory
:param project: title of the project
:param data_path: path to input data sample used for notebook
:param version: Version of the pycalibration tool
"""
module_path = "{}".format(path.abspath(path.dirname(__file__)))
with open('{}/titlepage.tmpl'.format(module_path)) as file_:
title_tmp = Template(file_.read())
with open("{}/titlepage.tex.txt".format(sphinx_path), "w+") as mf:
mf.write(dedent(title_tmp.render(project=tex_escape(project),
data_path=tex_escape(data_path),
version=tex_escape(version))))
def tex_escape(text):
"""
Escape latex special characters found in the text
:param text: a plain text message
:return: the message escaped to appear correctly in LaTeX
"""
conv = {
'&': r'\&',
'%': r'\%',
'$': r'\$',
'#': r'\#',
'_': r'\_',
'{': r'\{',
'}': r'\}',
'~': r'\textasciitilde{}',
'^': r'\^{}',
'\\': r'\textbackslash{}',
'<': r'\textless{}',
'>': r'\textgreater{}',
}
key_list = sorted(conv.keys(), key=lambda item: - len(item))
regex = re.compile('|'.join(re.escape(str(key)) for key in key_list))
return regex.sub(lambda match: conv[match.group()], text)
def finalize(joblist, finaljob, run_path, out_path, project, calibration,
author, version, report_to, data_path='Unknown'):
print("Waiting on jobs to finish: {}".format(joblist))
while True:
found_jobs = set()
output = check_output(['squeue']).decode('utf8')
for line in output.split("\n"):
for job in joblist:
if str(job) in line:
found_jobs.add(job)
if len(found_jobs) == 0:
break
sleep(10)
prepare_plots(run_path)
make_timing_summary(run_path, joblist + [str(finaljob)])
sphinx_path = combine_report(run_path, calibration)
make_titlepage(sphinx_path, project, data_path, version)
make_report(sphinx_path, run_path, out_path, project, author, version,
report_to)
cal_tools/cal_tools/xfel.pdf → xfel_calibrate/xfel.pdf
View file @ bc78a342
File moved

Legal Notice | Contacts