finalize
atoi(text)
¶
Convert string to integer is possible
:param text: string to be converted :return: integer value or input string
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
get_job_info(jobs, fmt)
¶
Returns list of job information from sacct
:param jobs: List of job names :param fmt: List of fields to query for each job (passed to sacct) Result ordered according to order of jobs given Order of fields in inner lists follows fmt
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
def get_job_info(jobs: List[str], fmt: List[str]) -> List[List[str]]:
"""Returns list of job information from sacct
:param jobs: List of job names
:param fmt: List of fields to query for each job (passed to sacct)
Result ordered according to order of jobs given
Order of fields in inner lists follows fmt
"""
if not jobs:
return [] # Skip calling sacct if not using Slurm
# will use JobID to match results to jobs (duplicate field in fmt is OK)
fmt_query = ",".join(["JobID"] + fmt)
sacct_out = check_output(["sacct", "--truncate", "--parsable2", "--noheader",
f"--jobs={','.join(jobs)}",
f"--format={fmt_query}"])
lines = sacct_out.decode().split("\n")
missing_info = ["not-found"] * len(fmt)
job_info = {job: missing_info for job in jobs}
for line in lines:
parts = line.split("|")
if parts[0] in job_info:
job_info[parts[0]] = parts[1:]
return [job_info[job] for job in jobs]
make_report(run_path, cal_work_dir, 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 cal_work_dir: Run path of the slurm job :param project: Project title :param author: Author of the notebook :param version: Version of the notebook :param report_to: report path tailed with report name
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
def make_report(run_path: Path, cal_work_dir: Path, project: str,
author: str, version: str, report_to: Path):
"""
Create calibration report (pdf file)
Automatically generated report document results, produced by
jupyter-notebooks.
:param run_path: Path to sphinx run directory
:param cal_work_dir: Run path of the slurm job
:param project: Project title
:param author: Author of the notebook
:param version: Version of the notebook
:param report_to: report path tailed with report name
"""
report_name = report_to.stem
report_dir = report_to.parent
try:
check_call([sys.executable, "-m", "sphinx.cmd.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-quickstart. Is sphinx installed?"
"Generated simple index.rst instead")
# quickbuild went well we need to edit the index.rst and conf.py files
module_path = Path(__file__).absolute().parent
conf_fn = run_path / "conf.py"
tmp_conf_fn = run_path / "conf.py.tmp"
conf = SourceFileLoader("conf", str(conf_fn)).load_module()
l_var = [v for v in dir(conf) if not v.startswith('__')]
with tmp_conf_fn.open("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))
conf_fn.unlink()
move(str(tmp_conf_fn), str(conf_fn))
direntries = list(run_path.iterdir())
lead_rstfiles = ['InputParameters.rst', 'timing_summary.rst']
# Order rst files based on the known order(lead_rstfiles).
# TODO: fix order somewhere else instead of munging filenames
def sort_key(f):
if f.name in lead_rstfiles:
return lead_rstfiles.index(f.name), f.name
elif "summary" in f.name.lower():
return len(lead_rstfiles), f.name
elif "precorrection" in f.name.lower():
return len(lead_rstfiles) + 1, f.name
else:
return len(lead_rstfiles) + 2, f.name
direntries.sort(key=sort_key)
files_to_handle = []
for entry in direntries:
if entry.is_file():
name, ext = entry.stem, entry.suffix
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 (run_path / "index.rst").open("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", f"SPHINXBUILD={sys.executable} -m sphinx",
"latexpdf"])
except CalledProcessError:
print("Failed to make pdf documentation")
print("Temp files will not be deleted and "
f"can be inspected at: {run_path}")
return
print(f"Moving report to final location: {report_dir}")
report_dir.mkdir(parents=True, exist_ok=True)
copy(run_path / "_build" / "latex" / f"{report_name}.pdf", report_dir)
# Remove folders with figures and sphinx files.
for tmp_subdir in cal_work_dir.iterdir():
if tmp_subdir.is_dir():
print(f"Removing temporary subdir: {tmp_subdir}")
rmtree(tmp_subdir)
make_timing_summary(cal_work_dir, job_times, job_time_fmt, pipeline_times)
¶
Create an rst file with timing summary of executed notebooks
:param cal_work_dir: Run path of the slurm job :param job_times: List of job information as returned by get_job_info :param job_time_fmt: List of headers to use for job_times :param pipeline_times: Dictionary of pipeline step -> timestamp
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
def make_timing_summary(cal_work_dir: Path, job_times: List[List[str]],
job_time_fmt: List[str], pipeline_times: Dict[str, datetime]):
"""
Create an rst file with timing summary of executed notebooks
:param cal_work_dir: Run path of the slurm job
:param job_times: List of job information as returned by get_job_info
:param job_time_fmt: List of headers to use for job_times
:param pipeline_times: Dictionary of pipeline step -> timestamp
"""
print('Prepare timing summary')
tmpl = Template('''
Runtime summary
===============
All timestamps are shown in local (Hamburg) time.
.. math::
{% for line in time_table %}
{{ line }}
{%- endfor %}
''')
job_tbl_tmpl = Template('''
.. math::
{% for line in job_table %}
{{ line }}
{%- endfor %}
''')
time_vals = [
[title, pipeline_times[k].astimezone(None).strftime("%Y-%m-%d %H:%M:%S")]
for (title, k) in [
["Time of Request", "request-time"],
["Job submission", "submission-time"],
["Report compilation", "report-compilation-time"],
]]
with (cal_work_dir / "timing_summary.rst").open("w+") as fd:
time_table = tabulate.tabulate(time_vals, tablefmt="latex",
headers=["Processing step", "Timestamp"])
fd.write(dedent(tmpl.render(time_table=time_table.split("\n"))))
if job_times:
job_table = tabulate.tabulate(job_times, tablefmt="latex",
headers=job_time_fmt)
fd.write(dedent(job_tbl_tmpl.render(job_table=job_table.split("\n"))))
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
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
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))))
natural_keys(text)
¶
Decompose string to list of integers and sub-strings
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
prepare_plots(cal_work_dir, 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 cal_work_dir: Run path of the slurm job :param threshold: Max svg file size (in bytes) to be converted to pdf
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
def prepare_plots(cal_work_dir: Path, threshold=1_000_000):
"""
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 cal_work_dir: 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')
for rst_file in cal_work_dir.glob("*.rst"):
for f_path in (cal_work_dir / f'{rst_file.stem}_files').glob('*.svg'):
if f_path.stat().st_size < threshold:
check_call(["svg2pdf", str(f_path)], shell=False)
new_ext = 'pdf'
else:
check_call(["convert", str(f_path),
str(f_path.with_suffix('.png'))], shell=False)
new_ext = 'png'
check_call([
"sed", "-i", f"s/{f_path.name}/{f_path.stem}.{new_ext}/g", rst_file
], shell=False)
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
Source code in /usr/src/app/checkouts/readthedocs.org/user_builds/european-xfel-offline-calibration/envs/latest/lib/python3.8/site-packages/xfel_calibrate/finalize.py
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)