From ae7aaee0ee6eb428e2673aa56fd2b65b6eeb7df0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ole=20Sch=C3=BCtt?= Date: Wed, 7 Apr 2021 16:42:51 +0200 Subject: [PATCH] Dashboard: Add Python type annotations --- tools/dashboard/generate_dashboard.py | 496 +++++++++++++++----------- 1 file changed, 279 insertions(+), 217 deletions(-) diff --git a/tools/dashboard/generate_dashboard.py b/tools/dashboard/generate_dashboard.py index 19d809139c..c244c34c86 100755 --- a/tools/dashboard/generate_dashboard.py +++ b/tools/dashboard/generate_dashboard.py @@ -5,6 +5,8 @@ # # author: Ole Schuett +from typing import Any, Literal, List, Dict, cast, Optional, ValuesView, NewType + import sys import os import smtplib @@ -17,73 +19,137 @@ from subprocess import check_output import traceback from os import path from pprint import pformat -from glob import glob import itertools -import configparser +from configparser import ConfigParser import pickle from pathlib import Path import requests import hashlib +import dataclasses +import argparse -import matplotlib as mpl +import matplotlib as mpl # type: ignore mpl.use("Agg") # change backend, to run without X11 -import matplotlib.pyplot as plt -from matplotlib.ticker import AutoMinorLocator +import matplotlib.pyplot as plt # type: ignore +from matplotlib.ticker import AutoMinorLocator # type: ignore # ====================================================================================== -class GitLog(list): - def __init__(self): +GitSha = NewType("GitSha", str) +ReportStatus = Literal["OK", "FAILED", "OUTDATED", "UNKNOWN"] + +# ====================================================================================== +class GitLog: + def __init__(self) -> None: + self.commits: List[Commit] = [] cmd = ["git", "log", "--pretty=format:%H%n%ct%n%an%n%ae%n%s%n%b<--separator-->"] outbytes = check_output(cmd) output = outbytes.decode("utf-8", errors="replace") for entry in output.split("<--separator-->")[:-1]: lines = entry.strip().split("\n") - commit = dict() - commit["git-sha"] = lines[0] - commit["date"] = datetime.fromtimestamp(float(lines[1])) - commit["author-name"] = lines[2] - commit["author-email"] = lines[3] - commit["msg"] = lines[4] - self.append(commit) + commit = Commit( + sha=GitSha(lines[0]), + date=datetime.fromtimestamp(float(lines[1])), + author_name=lines[2], + author_email=lines[3], + message=lines[4], + ) + self.commits.append(commit) # git-log outputs entries from new to old. - self.index = {c["git-sha"]: i for i, c in enumerate(self)} + self.index_by_sha = {c.sha: i for i, c in enumerate(self.commits)} # ====================================================================================== -def main(): - if len(sys.argv) < 4: - usage = " [--send-emails]" - print(f"Usage: generate_dashboard.py {usage}") - sys.exit(1) +@dataclasses.dataclass +class Commit: + sha: GitSha + author_name: str + author_email: str + date: datetime + message: str - config_fn, status_fn, outdir = sys.argv[1:4] - outdir = outdir.strip("/") - assert path.exists(config_fn) - global send_emails - if len(sys.argv) == 5: - assert sys.argv[4] == "--send-emails" - send_emails = True - else: - send_emails = False +# ====================================================================================== +@dataclasses.dataclass +class PlotCurve: + label: str + x: List[float] = dataclasses.field(default_factory=list) + y: List[float] = dataclasses.field(default_factory=list) + yerr: List[float] = dataclasses.field(default_factory=list) - config = configparser.ConfigParser() - config.read(config_fn) + +# ====================================================================================== +class Plot: + def __init__(self, name: str, title: str, ylabel: str, **_: Any): + self.name = name + self.title = title + self.ylabel = ylabel + self.curves_by_name: Dict[str, PlotCurve] = {} + + @property + def curves(self) -> ValuesView[PlotCurve]: + return self.curves_by_name.values() + + +# ====================================================================================== +class PlotPoint: + def __init__( + self, plot: str, name: str, label: str, y: float, yerr: float, **_: Any + ): + self.plot_name = plot + self.curve_name = name + self.curve_label = label + self.y = y + self.yerr = yerr + + +# ====================================================================================== +@dataclasses.dataclass +class Report: + status: ReportStatus + summary: str + sha: Optional[GitSha] = None + plots: List[Plot] = dataclasses.field(default_factory=list) + plot_points: List[PlotPoint] = dataclasses.field(default_factory=list) + archive_path: Optional[str] = None + + +# ====================================================================================== +@dataclasses.dataclass +class Status: + notified: bool = False + last_ok: Optional[GitSha] = None + + +# ====================================================================================== +def main() -> None: + parser = argparse.ArgumentParser(description="Generates the CP2K Dashboard.") + parser.add_argument("config_file", type=Path) + parser.add_argument("status_file", type=Path) + parser.add_argument("output_dir", type=Path) + parser.add_argument("--send-emails", action="store_true") + args = parser.parse_args() + + assert args.config_file.exists() + config = ConfigParser() + config.read(args.config_file) log = GitLog() # Reads history from local git repo. - gen_frontpage(config, log, status_fn, outdir) - gen_archive(config, log, outdir) - gen_url_list(config, outdir) + gen_frontpage(config, log, args.status_file, args.output_dir, args.send_emails) + gen_archive(config, log, args.output_dir) + gen_url_list(config, args.output_dir) # ====================================================================================== -def gen_frontpage(config, log, status_fn, outdir): +def gen_frontpage( + config: ConfigParser, log: GitLog, status_fn: Path, outdir: Path, send_emails: bool +) -> None: + + status: Dict[str, Status] = {} if path.exists(status_fn): - status = eval(open(status_fn, encoding="utf8").read()) - else: - status = dict() + with open(status_fn, "rb") as f: + status = pickle.load(f) output = html_header(title="CP2K Dashboard") output += '
\n' @@ -94,12 +160,9 @@ def gen_frontpage(config, log, status_fn, outdir): output += "NameHostStatus" output += "CommitSummaryLast OK\n\n" - def get_sortkey(s): - return config.getint(s, "sortkey") - now = datetime.utcnow().replace(microsecond=0) - for s in sorted(config.sections(), key=get_sortkey): + for s in sorted(config.sections(), key=lambda s: config.getint(s, "sortkey")): print("Working on summary entry of: " + s) name = config.get(s, "name") host = config.get(s, "host") @@ -109,46 +172,46 @@ def gen_frontpage(config, log, status_fn, outdir): # find latest commit that should have been tested by now threshold = now - timedelta(hours=timeout) - ages_beyond_threshold = [i for i, c in enumerate(log) if c["date"] < threshold] - threshold_age = ages_beyond_threshold[0] + threshold_age = next(i for i, c in enumerate(log.commits) if c.date < threshold) # get and parse report report_txt = retrieve_report(report_url) report = parse_report(report_txt, log) if s not in status: - status[s] = {"last_ok": None, "notified": False} + status[s] = Status() - if report["status"] == "OK": - status[s]["last_ok"] = report["git-sha"] - status[s]["notified"] = False - elif do_notify and not status[s]["notified"]: - send_notification(report, status[s]["last_ok"], log, name, s) - status[s]["notified"] = True + if report.status == "OK": + status[s].last_ok = report.sha + status[s].notified = False + elif do_notify and not status[s].notified: + send_notification(report, status[s].last_ok, log, name, s, send_emails) + status[s].notified = True - if report["git-sha"]: - age = log.index[report["git-sha"]] + if report.sha: + age = log.index_by_sha[report.sha] if age > threshold_age: - report["status"] = "OUTDATED" - elif report["status"] in ("OK", "FAILED"): + report.status = "OUTDATED" + elif report.status in ("OK", "FAILED"): # store only useful and fresh reports, prevents overwriting archive + assert report_txt store_report(report, report_txt, s, outdir) - uptodate = report["git-sha"] == log[0]["git-sha"] # report from latest commit? + up_to_date = report.sha == log.commits[0].sha # report from latest commit? output += '' output += f'{name}' output += f'{host}' - output += status_cell(report["status"], report_url, uptodate) + output += status_cell(report.status, report_url, up_to_date) # Commit - output += commit_cell(report["git-sha"], log) + output += commit_cell(report.sha, log) # Summary - output += f'{report["summary"]}' + output += f'{report.summary}' # Last OK - if report["status"] != "OK": - output += commit_cell(status[s]["last_ok"], log) + if report.status != "OK": + output += commit_cell(status[s].last_ok, log) else: output += "" @@ -157,25 +220,27 @@ def gen_frontpage(config, log, status_fn, outdir): output += "\n" output += '
\n' # complete flex-container output += html_footer() - write_file(f"{outdir}/index.html", output) - write_file(status_fn, pformat(status)) + write_file(outdir / "index.html", output) + with open(status_fn, "wb") as f: + pickle.dump(status, f) # ====================================================================================== -def gen_archive(config, log, outdir): +def gen_archive(config: ConfigParser, log: GitLog, outdir: Path) -> None: for s in config.sections(): print(f"Working on archive page of: {s}") name = config.get(s, "name") info_url = config.get(s, "info_url", fallback=None) - archive_files = glob(f"{outdir}/archive/{s}/commit_*.txt.gz") + archive_dir = outdir / f"archive/{s}" + archive_files = archive_dir.glob("commit_*.txt.gz") # read cache - cache_fn = f"{outdir}/archive/{s}/reports.cache" - if not path.exists(cache_fn): - reports_cache = dict() - else: - reports_cache = pickle.load(open(cache_fn, "rb")) + reports_cache: Dict[Path, Report] = {} + cache_fn = f"{archive_dir}/reports.cache" + if path.exists(cache_fn): + with open(cache_fn, "rb") as f: + reports_cache = pickle.load(f) cache_age = path.getmtime(cache_fn) # remove outdated cache entries reports_cache = { @@ -183,42 +248,45 @@ def gen_archive(config, log, outdir): } # read all archived reports - archive_reports = dict() + archive_reports: Dict[GitSha, Report] = {} for fn in archive_files: + report: Report if fn in reports_cache: report = reports_cache[fn] else: - with gzip.open(fn, "rb") as f: - report_txt = f.read().decode("utf-8", errors="replace") - report = parse_report(report_txt, log) - report["url"] = path.basename(fn)[:-3] + with open(fn, "rb") as f: + content = gzip.decompress(f.read()) + report = parse_report(content.decode("utf-8", errors="replace"), log) + report.archive_path = path.basename(fn)[:-3] reports_cache[fn] = report - sha = report["git-sha"] - assert sha not in archive_reports - archive_reports[sha] = report + assert report.sha and report.sha not in archive_reports + archive_reports[report.sha] = report # write cache if reports_cache: - pickle.dump(reports_cache, open(cache_fn, "wb")) + with open(cache_fn, "wb") as f: + pickle.dump(reports_cache, f) # loop over all relevant commits all_url_rows = [] all_html_rows = [] - max_age = 1 + max([-1] + [log.index[sha] for sha in archive_reports.keys()]) - for commit in log[:max_age]: - sha = commit["git-sha"] + max_age = 1 + max([-1] + [log.index_by_sha[sha] for sha in archive_reports]) + for commit in log.commits[:max_age]: + sha = commit.sha html_row = "" html_row += commit_cell(sha, log) if sha in archive_reports: report = archive_reports[sha] - html_row += status_cell(report["status"], report["url"]) - html_row += f'{report["summary"]}' - url_row = f'https://dashboard.cp2k.org/archive/{s}/{report["url"]}.gz\n' + assert report.archive_path + html_row += status_cell(report.status, report.archive_path) + html_row += f'{report.summary}' + archive_base_url = "https://dashboard.cp2k.org/archive/{s}" + url_row = f"{archive_base_url}/{report.archive_path}.gz\n" else: html_row += 2 * "" url_row = "" - html_row += f'{html.escape(commit["author-name"])}' - html_row += f'{html.escape(commit["msg"])}' + html_row += f'{html.escape(commit.author_name)}' + html_row += f'{html.escape(commit.message)}' html_row += "\n\n" all_html_rows.append(html_row) all_url_rows.append(url_row) @@ -229,7 +297,6 @@ def gen_archive(config, log, outdir): html_out_postfix = "index_full.html" urls_out_postfix = "list_full.txt" toggle_link = '

View recent archive

' - max_age = None # output all else: html_out_postfix = "index.html" urls_out_postfix = "list_recent.txt" @@ -241,9 +308,7 @@ def gen_archive(config, log, outdir): output += '

Go back to main page

\n' if info_url: output += f'

Get more information

\n' - output += gen_plots( - archive_reports, log, f"{outdir}/archive/{s}", full_archive - ) + output += gen_plots(archive_reports, log, archive_dir, full_archive) output += toggle_link output += '\n' output += "" @@ -252,106 +317,105 @@ def gen_archive(config, log, outdir): output += "
CommitStatusSummary
\n" output += toggle_link output += html_footer() - write_file(f"{outdir}/archive/{s}/{html_out_postfix}", output) + write_file(archive_dir / html_out_postfix, output) url_list = "".join(all_url_rows[:max_age]) - write_file(f"{outdir}/archive/{s}/{urls_out_postfix}", url_list) + write_file(archive_dir / urls_out_postfix, url_list) # ====================================================================================== -def gen_url_list(config, outdir): +def gen_url_list(config: ConfigParser, outdir: Path) -> None: print("Working on url lists.") for postfix in ("list_full.txt", "list_recent.txt"): url_list = "" for s in config.sections(): - fn = f"{outdir}/archive/{s}/{postfix}" + fn = outdir / f"archive/{s}/{postfix}" if not path.exists(fn): continue - url_list += open(fn, encoding="utf8").read() - write_file(f"{outdir}/archive/{postfix}", url_list) + with open(fn, encoding="utf8") as f: + url_list += f.read() + write_file(outdir / f"archive/{postfix}", url_list) # ====================================================================================== -def gen_plots(archive_reports, log, outdir, full_archive): +def gen_plots( + archive_reports: Dict[GitSha, Report], log: GitLog, outdir: Path, full_archive: bool +) -> str: - ordered_shas = [commit["git-sha"] for commit in log] + ordered_shas = [c.sha for c in log.commits] ordered_reports = [archive_reports[s] for s in ordered_shas if s in archive_reports] # collect plot data - plots = {} + plots_by_name = {} for report in ordered_reports: - for p in report["plots"]: - if p["name"] not in plots.keys(): - new_plot = {"curves": {}, "title": p["title"], "ylabel": p["ylabel"]} - plots[p["name"]] = new_plot - for pp in report["plotpoints"]: - p = plots[pp["plot"]] - if pp["name"] not in p["curves"].keys(): - new_curve = {"x": [], "y": [], "yerr": [], "label": pp["label"]} - p["curves"][pp["name"]] = new_curve - c = p["curves"][pp["name"]] - age = log.index[report["git-sha"]] - c["x"].append(-age) - c["y"].append(pp["y"]) - c["yerr"].append(pp["yerr"]) + for p in report.plots: + if p.name not in plots_by_name: + plots_by_name[p.name] = p + for pp in report.plot_points: + plot = plots_by_name[pp.plot_name] + cname = pp.curve_name + if cname not in plot.curves_by_name: + plot.curves_by_name[cname] = PlotCurve(label=pp.curve_label) + assert report.sha + age = log.index_by_sha[report.sha] + plot.curves_by_name[cname].x.append(-age) + plot.curves_by_name[cname].y.append(pp.y) + plot.curves_by_name[cname].yerr.append(pp.yerr) - max_age_full = max([-1] + [log.index[sha] for sha in archive_reports.keys()]) + max_age_full = max([-1] + [log.index_by_sha[sha] for sha in archive_reports]) max_age = max_age_full if full_archive else 100 output = "" - for pname, p in plots.items(): + for pname, plot in plots_by_name.items(): print(f"Working on plot: {pname}") fn = f"{pname}_full" if full_archive else pname - title = p["title"] - make_plot_data(f"{outdir}/{fn}", title, p["curves"], max_age, log) - make_plot_image( - f"{outdir}/{fn}", title, p["ylabel"], p["curves"], max_age, full_archive - ) - output += f'{title}\n' + make_plot_data(outdir / Path(f"{fn}.txt"), plot, max_age, log) + make_plot_image(outdir / Path(f"{fn}.png"), plot, max_age, full_archive) + output += f'{plot.title}\n' return output # ====================================================================================== -def make_plot_data(fn, title, curves, max_age, log): - output = f"# {title}\n" - header = ["age", "commit"] + list(curves.keys()) - output += "# " + "\t".join(header) + "\n" +def make_plot_data(fn: Path, plot: Plot, max_age: int, log: GitLog) -> None: + output = f"# {plot.title}\n" + headers = [f"{c.label:>6}" for c in plot.curves] + output += "# " + "\t".join(["age", "commit"] + headers) + "\n" for age in range(max_age + 1): values = [] - for label, c in curves.items(): - if -age in c["x"]: - i = c["x"].index(-age) - value = "{:.1f}".format(c["y"][i]) + for curve, header in zip(plot.curves, headers): + if -age in curve.x: + i = curve.x.index(-age) + value = "{:.1f}".format(curve.y[i]) else: value = "?" - values.append(value.rjust(len(label))) + values.append(value.rjust(len(header))) if any([v.strip() != "?" for v in values]): - index_colums = [f"{-age:5d}", log[age]["git-sha"][:6]] + index_colums = [f"{-age:5d}", log.commits[age].sha[:6]] output += "\t".join(index_colums + values) + "\n" - write_file(f"{fn}.txt", output) + write_file(fn, output) # ====================================================================================== -def make_plot_image(fn, title, ylabel, curves, max_age, full_archive): +def make_plot_image(fn: Path, plot: Plot, max_age: int, full_archive: bool) -> None: # Setup figure. fig = plt.figure(figsize=(12, 4)) fig.subplots_adjust(bottom=0.18, left=0.06, right=0.70) - fig.suptitle(title, fontsize=14, fontweight="bold", x=0.4) + fig.suptitle(plot.title, fontsize=14, fontweight="bold", x=0.4) ax = fig.add_subplot(111) ax.set_xlabel("Commit Age") - ax.set_ylabel(ylabel) + ax.set_ylabel(plot.ylabel) markers = itertools.cycle("os>^*") # Draw curves. - for cname, c in curves.items(): + for curve in plot.curves: if full_archive: - ax.plot(c["x"], c["y"], label=c["label"], linewidth=2) # less crowded + ax.plot(curve.x, curve.y, label=curve.label, linewidth=2) # less crowded else: style = dict(marker=next(markers), linewidth=2, markersize=6) - ax.errorbar(c["x"], c["y"], yerr=c["yerr"], label=c["label"], **style) + ax.errorbar(curve.x, curve.y, yerr=curve.yerr, label=curve.label, **style) ax.set_xlim(-max_age - 1, 0) ax.xaxis.set_minor_locator(AutoMinorLocator()) @@ -361,8 +425,8 @@ def make_plot_image(fn, title, ylabel, curves, max_age, full_archive): # Determine y-range such that all curves are visible while ignoring outlayers. visibles = [] - for c in curves.values(): - ys = [y for x, y in zip(c["x"], c["y"]) if x >= -max_age] # visible y-values + for curve in plot.curves: + ys = [y for x, y in zip(curve.x, curve.y) if x >= -max_age] # visible y-values visibles += [ys] if ys else [] # ignore completely invisible curves if not visibles: print("Warning: Found no visible plot curve.") @@ -376,20 +440,28 @@ def make_plot_image(fn, title, ylabel, curves, max_age, full_archive): ax.set_ylim(0.98 * ymin, min(1.02 * ymax, 1.3 * ymax2)) # ignore outlayers # Save figure to file. - fig.savefig(fn + ".png") + fig.savefig(fn) plt.close(fig) # ====================================================================================== -def send_notification(report, last_ok, log, name, s): - idx_end = log.index[report["git-sha"]] if (report["git-sha"]) else 0 +def send_notification( + report: Report, + last_ok: Optional[GitSha], + log: GitLog, + name: str, + s: str, + send_emails: bool, +) -> None: + + idx_end = log.index_by_sha[report.sha] if report.sha else 0 if not last_ok: return # we don't know when this started - idx_last_ok = log.index[last_ok] + idx_last_ok = log.index_by_sha[last_ok] if idx_end == idx_last_ok: return # probably a flapping tester - emails = set([log[i]["author-email"] for i in range(idx_end, idx_last_ok)]) - emails = [e for e in emails if "noreply" not in e] + emails_raw = set(c.author_email for c in log.commits[idx_end:idx_last_ok]) + emails = [e for e in emails_raw if "noreply" not in e] emails_str = ", ".join(emails) if not emails: return # no author emails found @@ -406,8 +478,8 @@ def send_notification(report, last_ok, log, name, s): msg_txt += "the dashboard has detected a problem" msg_txt += " that one of your recent commits might have introduced.\n\n" msg_txt += f" test name: {name}\n" - msg_txt += f" report state: {report['status']}\n" - msg_txt += f" report summary: {report['summary']}\n" + msg_txt += f" report state: {report.status}\n" + msg_txt += f" report summary: {report.summary}\n" msg_txt += f" last OK commit: {last_ok[:7]}\n\n" msg_txt += "For more information visit:\n" msg_txt += f" https://dashboard.cp2k.org/archive/{s}/index.html \n\n" @@ -425,7 +497,7 @@ def send_notification(report, last_ok, log, name, s): # ====================================================================================== -def html_header(title): +def html_header(title: str) -> str: output = '\n' output += "\n" output += '\n' @@ -493,7 +565,7 @@ def html_header(title): # ====================================================================================== -def html_linkbox(): +def html_linkbox() -> str: output = '
\n' output += "

More...

\n" output += 'Regtest Survey
\n' @@ -504,27 +576,26 @@ def html_linkbox(): # ====================================================================================== -def html_gitbox(log): +def html_gitbox(log: GitLog) -> str: now = datetime.utcnow() output = '
\n' output += "

Recent Commits

\n" - for commit in log[0:10]: - url = "https://github.com/cp2k/cp2k/commit/" + commit["git-sha"] - msg = commit["msg"] - autor_name = html.escape(commit["author-name"]) - msg = commit["msg"] if len(commit["msg"]) < 28 else commit["msg"][:26] + "..." - title = html.escape(commit["msg"]) - output += f'

{html.escape(msg)}
\n' - delta = now - commit["date"] + for commit in log.commits[0:10]: + url = "https://github.com/cp2k/cp2k/commit/" + commit.sha + autor_name = html.escape(commit.author_name) + title = html.escape(commit.message) + cm = commit.message if len(commit.message) < 28 else commit.message[:26] + "..." + output += f'

{html.escape(cm)}
\n' + delta = now - commit.date age = delta.days * 24.0 + delta.seconds / 3600.0 - output += "git:" + commit["git-sha"][:7] + output += "git:" + commit.sha[:7] output += f"
\n{autor_name} {age:.1f}h ago.

\n" output += "
\n" return output # ====================================================================================== -def html_footer(): +def html_footer() -> str: now = datetime.utcnow().replace(microsecond=0) output = f"

Page last updated: {now.isoformat()}

\n" output += "" @@ -532,46 +603,48 @@ def html_footer(): # ====================================================================================== -def write_file(fn, content, gz=False): - d = path.dirname(fn) - if len(d) > 0 and not path.exists(d): +def write_file(fn: Path, content: str, gz: bool = False) -> None: + d = fn.parent + if not d.exists(): os.makedirs(d) - print("Created dir: " + d) - if path.exists(fn): - old_bytes = gzip.open(fn, "rb").read() if (gz) else open(fn, "rb").read() + print(f"Created dir: {d}") + if fn.exists(): + with open(fn, "rb") as f: + old_bytes = gzip.decompress(f.read()) if gz else f.read() old_content = old_bytes.decode("utf-8", errors="replace") if old_content == content: - print("File did not change: " + fn) + print(f"File did not change: {fn}") return - f = gzip.open(fn, "wb") if (gz) else open(fn, "wb") - f.write(content.encode("utf-8")) - f.close() - print("Wrote: " + fn) + + encoded_content = content.encode("utf-8") + with open(fn, "wb") as f: + f.write(gzip.compress(encoded_content) if gz else encoded_content) + print(f"Wrote: {fn}") # ====================================================================================== -def status_cell(status, report_url, uptodate=True): +def status_cell(status: ReportStatus, report_url: str, up_to_date: bool = True) -> str: if status == "OK": - bgcolor = "#00FF00" if (uptodate) else "#8CE18C" + bgcolor = "#00FF00" if (up_to_date) else "#8CE18C" elif status == "FAILED": - bgcolor = "#FF0000" if (uptodate) else "#E18C8C" + bgcolor = "#FF0000" if (up_to_date) else "#E18C8C" else: bgcolor = "#d3d3d3" return f'{status}' # ====================================================================================== -def commit_cell(git_sha, log): - if git_sha is None: +def commit_cell(sha: Optional[GitSha], log: GitLog) -> str: + if sha is None: return "N/A" - idx = log.index[git_sha] - commit = log[idx] - git_url = "https://github.com/cp2k/cp2k/commit/" + git_sha - return f'{git_sha[:7]} ({-idx})' + idx = log.index_by_sha[sha] + commit = log.commits[idx] + github_url = "https://github.com/cp2k/cp2k/commit/" + sha + return f'{sha[:7]} ({-idx})' # ====================================================================================== -def retrieve_report(url): +def retrieve_report(url: str) -> Optional[str]: try: # see if we have a cached entry h = hashlib.md5(url.encode("utf8")).hexdigest() @@ -596,57 +669,46 @@ def retrieve_report(url): return r.text except: - print(traceback.print_exc()) + traceback.print_exc() return None # ====================================================================================== -def store_report(report, report_txt, section, outdir): - fn = f'{outdir}/archive/{section}/commit_{report["git-sha"]}.txt.gz' +def store_report(report: Report, report_txt: str, section: str, outdir: Path) -> None: + fn = outdir / f"archive/{section}/commit_{report.sha}.txt.gz" write_file(fn, report_txt, gz=True) # ====================================================================================== -def parse_report(report_txt, log): +def parse_report(report_txt: Optional[str], log: GitLog) -> Report: if report_txt is None: - return { - "status": "UNKNOWN", - "summary": "Error while retrieving report.", - "git-sha": None, - } + return Report(status="UNKNOWN", summary="Error while retrieving report.") try: - report = dict() - git_sha = re.search("(^|\n)CommitSHA: (\w{40})\n", report_txt).group(2) - report["git-sha"] = git_sha - report["summary"] = re.findall("(^|[\n\r])Summary: (.+)\n", report_txt)[-1][1] - report["status"] = re.findall("(^|\n)Status: (.+)\n", report_txt)[-1][1] - all_plots = re.findall("(^|\n)Plot: (.+)(?=\n)", report_txt) - report["plots"] = [eval(f"dict({m[1]})") for m in all_plots] - all_plotpoints = re.findall("(^|\n)PlotPoint: (.+)(?=\n)", report_txt) - report["plotpoints"] = [eval(f"dict({m[1]})") for m in all_plotpoints] + m = re.search("(^|\n)CommitSHA: (\w{40})\n", report_txt) + sha = GitSha(m.group(2)) if m else None + summary = re.findall("(^|[\n\r])Summary: (.+)\n", report_txt)[-1][1] + status = re.findall("(^|\n)Status: (.+)\n", report_txt)[-1][1] + plot_matches = re.findall("(^|\n)Plot: (.+)(?=\n)", report_txt) + plots = [cast(Plot, eval(f"Plot({m[1]})")) for m in plot_matches] + point_matches = re.findall("(^|\n)PlotPoint: (.+)(?=\n)", report_txt) + points = [cast(PlotPoint, eval(f"PlotPoint({m[1]})")) for m in point_matches] # Check that every plot has at least one PlotPoint - for plot in report["plots"]: - points = [pp for pp in report["plotpoints"] if pp["plot"] == plot["name"]] - if not points: - report["status"] = "FAILED" - report["summary"] = f"Plot {plot['name']} has no PlotPoints." + for plot in plots: + if not any([p.plot_name == plot.name for p in points]): + return Report("FAILED", f"Plot {plot.name} has no PlotPoints.") # Check that CommitSHA belongs to the master branch. - if report["git-sha"] not in log.index: - report["git-sha"] = None - report["status"] = "FAILED" - report["summary"] = "Unknown CommitSHA." + if sha not in log.index_by_sha: + return Report("FAILED", "Unknown CommitSHA.") - return report + return Report(status, summary, sha=sha, plots=plots, plot_points=points) except: - print(traceback.print_exc()) - return { - "status": "UNKNOWN", - "summary": "Error while parsing report.", - "git-sha": None, - } + traceback.print_exc() + return Report("UNKNOWN", "Error while parsing report.") # ====================================================================================== main() + +# EOF