from datetime import datetime
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, request_time, submission_time):
"""
Create an rst file with timing summary of executed notebooks
:param run_path: Run path of the slurm job
:param joblist: List of slurm jobs
:param request_time: Timestamp of notebook request
:param submission_time: Timestamp for slurm jobs being submitted
"""
print('Prepare timing summary')
run_path = path.abspath(run_path)
pars_vals = []
pars = 'JobID,Start,End,Elapsed,Suspended,State'
pars_name = pars.split(',')
for job in joblist:
out = check_output(['sacct', '-T', '-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 time_table %}
{{ line }}
{%- endfor %}
.. math::
{% for line in job_table %}
{{ line }}
{%- endfor %}
''')
time_vals = [['Time of Request', request_time],
['Job submission', submission_time],
['Report compilation',
datetime.now().strftime('%Y-%m-%dT%H:%M:%S')]]
with open("{}/timing_summary.rst".format(run_path), "w+") as gfile:
if len(pars_vals) > 0:
job_table = tabulate.tabulate(pars_vals, tablefmt='latex',
headers=pars_name)
time_table = tabulate.tabulate(time_vals, tablefmt='latex',
headers=['Processing step',
'Timestamp'])
gfile.write(dedent(tmpl.render(job_table=job_table.split('\n'),
time_table=time_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)
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 in lead_rstfiles:
return lead_rstfiles.index(f), f
elif "summary" in f.lower():
return len(lead_rstfiles), f
elif "precorrection" in f.lower():
return len(lead_rstfiles) + 1, f
else:
return len(lead_rstfiles) + 2, f
lead_rstfiles.sort(key=sort_key)
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))
makedirs(out_path, exist_ok=True)
copy('{}/_build/latex/{}.pdf'.format(run_path, report_name), out_path)
temp_dirs = glob(f'{tmp_path}/*/')
# Remove folders with figures and sphinx files.
print(f"Removing directories [{temp_dirs}] in temp folder: {tmp_path}")
for dtmp in temp_dirs:
rmtree(f'{dtmp}/')
# Archiving files in slurm_tmp
if os.path.isfile(f'{out_path}/retrieved_constants.yml'):
move(f'{out_path}/retrieved_constants.yml',
f"{tmp_path}")
# Moving temporary files to out-folder after successful execution
# This helps in keeping elements needed for re-producibility.
print(f"Moving temporary files to final location"
f": {out_path}/{os.path.basename(tmp_path)} with name: "
f"slurm_out_{report_name}")
move(tmp_path, f"{out_path}/slurm_out_{report_name}")
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',
request_time='', submission_time=''):
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)], request_time,
submission_time)
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)