Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602#!/usr/bin/env python3"""Posts one message to the community Discord's #nev channel when a feature job lands, orwhen any other landing moves a web feature to partial or complete: what landed, whichfeatures moved, and what the work behind it took.
bench/feature-announce.py --events <file> [--result <file>] [--track <name>] [--ledger <file>] [--since <unix time>] [--now <unix time>] [--inventory <file>] [--root <dir>] [--snapshot <file>] [--dry-run]
--events the event lines the landing's web-features update appended (docs/web-features-events.jsonl; bench/after-land.sh passes only the new ones). Only "feature-status" events that move a feature forward to partial or complete count (from none, unknown or new to partial or complete, or from partial to complete); complete -> partial is a regression and is not announced. Without --result, none of them: one line, nothing posted, exit 0.--result the landed job's result.json (jobs/src/result.rs). With it the landing always posts, once: the job's name and its spec's title line (specs/<name>.md), each generated function it admitted (the *.provenance.json files its stages read: the model, and cases passed out of cases run against the three browsers), the job's run time (first round's start to last round's end), and the map: the features that moved, or "map: no status change" with the reason (bench/expected-features-check.py's: already partial or complete, no map rule, or a job that declares no feature). An unreadable result: one line, and the rule without --result applies.--root the checkout the spec and provenance files are read from (default: this one).--snapshot the web-features map for the reason line (default <root>/docs/web-features-map.json).--track the landed track (its worktree directory name), whose ledger gives the numbers. Without it the post names the features only.--ledger the track's ledger file (bench/ledger.py). Otherwise the script looks for <track>.jsonl in NEV_LEDGER_DIR, then in $NEV_LANE_SHARED/ledger where that is mounted, then asks a build lane for it over ssh (the lane is found the way bench/on-lane.sh finds it). No ledger: the ledger lines in --result's `ledger`, if any; else the post carries no model numbers.--since the time the track's work began (its first commit); the wall time runs from the earlier of this and the first ledger entry to --now (default: now).--inventory the rig inventory (default bench/rig-inventory.local.json, gitignored): the cards serving each model's pool and their watts, for GPU-hours and energy. Without it, or for a model it does not list, GPU time is left out.--dry-run print the JSON body instead of posting.
The post goes to POST <NEV_ANNOUNCE_URL>/announce with "Authorization: Bearer<NEV_ANNOUNCE_TOKEN>", both read from the environment or, failing that, from thecheckout's .env (only those two keys are read from it; a key set in the environment, evento nothing, wins over the file). Either missing or empty: one line sayingthe post was skipped, exit 0. A failed post is one line and exit 1; the caller decideswhether that matters (bench/after-land.sh reports it and carries on). Neither the addressnor the token is ever printed. Body: {title, description, fields: [{name, value,inline}], footer}, kept within Discord's embed limits (title 256, description 4096, 25fields, field value 1024, footer 2048, 6000 in all).
Numbers in the post: - job run (with --result): the job's rounds, first start to last end; - per model: candidates drawn, compiled, best score, completion tokens; - wall time: first ledger entry (or --since, whichever is earlier) to the landing; left out when the job run is shown; - CPU-hours: the lane CPU-seconds the ledger recorded; - GPU-hours: each model's draw seconds times the cards serving its pool; - energy (an estimate): GPU-hours times each pool's card watts, plus CPU-hours times a per-CPU figure (`lane_watts_per_cpu`, else the verification tier's machine watts spread over its cores)."""
import argparseimport jsonimport osimport subprocessimport sysimport timeimport urllib.errorimport urllib.requestfrom pathlib import Path
BENCH_DIR = Path(__file__).resolve().parentROOT = BENCH_DIR.parent
LIMITS = {"title": 256, "description": 4096, "fields": 25, "field_name": 256, "field_value": 1024, "footer": 2048, "total": 6000}MAX_MODEL_FIELDS = 12MAX_FEATURE_LINES = 30# About what boiling a full kettle (1.7 l from tap temperature) takes.KETTLE_WH = 180RANK = {"partial": 1, "complete": 2}STATUS_WORDS = {"complete": "now supported", "partial": "now partly supported"}
def say(msg): print(f"announce: {msg}", file=sys.stderr)
# ---------------------------------------------------------------- inputs
def read_events(path): """The features this landing moved to partial or complete, in first-seen order; a feature that appears twice keeps its last move.""" moved = {} with open(path) as f: for line in f: line = line.strip() if not line: continue try: e = json.loads(line) except ValueError: continue if not isinstance(e, dict) or e.get("type") != "feature-status": continue to, frm = e.get("to"), e.get("from") # Forward moves only: complete -> partial is a regression, not news. if RANK.get(to, 0) > RANK.get(frm, 0) and isinstance(e.get("feature"), str): moved[e["feature"]] = {"feature": e["feature"], "name": e.get("name") or e["feature"], "from": frm, "to": to} return list(moved.values())
def parse_ledger(text): out = [] for line in text.splitlines(): try: r = json.loads(line) except ValueError: continue if isinstance(r, dict) and r.get("format") == 1: out.append(r) return out
def env_file_value(key): """One KEY=VALUE from the checkout's .env, one layer of quotes stripped; never sourced.""" env = ROOT / ".env" try: lines = env.read_text().splitlines() except OSError: return "" for line in lines: if line.startswith(key + "="): v = line[len(key) + 1:].strip() if len(v) >= 2 and v[0] == v[-1] and v[0] in "\"'": v = v[1:-1] return v return ""
def setting(key): """The environment's value when the key is set there at all (an empty value included, so `NEV_ANNOUNCE_URL= ...` turns posting off), else the checkout's .env.""" if key in os.environ: return os.environ[key] return env_file_value(key)
def lane_ssh(): """(target, options) of the first build lane, found as bench/on-lane.sh finds it.""" target = setting("NEV_LANE_SSH") or setting("GENERA_LANE_SSH") opts = setting("NEV_LANE_SSH_OPTS") or setting("GENERA_LANE_SSH_OPTS") if not target: try: cfg = json.loads((BENCH_DIR / "lane.local.json").read_text()) except (OSError, ValueError): cfg = {} lanes = cfg.get("lanes") if isinstance(cfg.get("lanes"), list) else [] first = lanes[0] if lanes and isinstance(lanes[0], dict) else cfg target = first.get("ssh", "") or "" opts = opts or first.get("ssh_opts", "") or "" return target, opts.split()
def ledger_from_lane(track): target, opts = lane_ssh() if not target: return None remote = ('for f in "$HOME/.lane-env" "$HOME/.lane-shared-env"; do ' '[ -f "$f" ] && . "$f" >/dev/null 2>&1; done; ' 'd="${NEV_LANE_SHARED:-${GENERA_LANE_SHARED:-}}"; ' f'[ -n "$d" ] && [ -f "$d/ledger/{track}.jsonl" ] && cat "$d/ledger/{track}.jsonl"') try: r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", *opts, target, remote], capture_output=True, text=True, timeout=30, stdin=subprocess.DEVNULL) except (OSError, subprocess.TimeoutExpired): return None return r.stdout if r.returncode == 0 else None
def find_ledger(track, path): if path: return Path(path).read_text() if not track: return None if not all(c.isalnum() or c in "._-" for c in track): say("the track name has characters a ledger file name cannot, so no ledger") return None for d in (os.environ.get("NEV_LEDGER_DIR", ""), os.path.join(os.environ.get("NEV_LANE_SHARED", ""), "ledger") if os.environ.get("NEV_LANE_SHARED") else ""): if d and (Path(d) / f"{track}.jsonl").is_file(): return (Path(d) / f"{track}.jsonl").read_text() if os.environ.get("NEV_LEDGER_DIR"): return None return ledger_from_lane(track)
def read_inventory(path): p = Path(path) if path else BENCH_DIR / "rig-inventory.local.json" try: return json.loads(p.read_text()) except (OSError, ValueError): return None
# ---------------------------------------------------------------- numbers
def totals(records): models = {} cpu = 0.0 cpu_known = False starts = [] for r in records: if isinstance(r.get("start"), (int, float)): starts.append(r["start"]) if isinstance(r.get("cpu_seconds"), (int, float)): cpu += r["cpu_seconds"] cpu_known = True for name, m in (r.get("models") or {}).items(): t = models.setdefault(name, {"candidates": 0, "compiled": 0, "completion_tokens": 0, "draw_seconds": 0.0, "best": None}) for k in ("candidates", "compiled", "completion_tokens", "draw_seconds"): v = m.get(k) if isinstance(v, (int, float)): t[k] += v b = m.get("best") if isinstance(b, dict) and b.get("run"): if t["best"] is None or b["passed"] * t["best"]["run"] > t["best"]["passed"] * b["run"]: t["best"] = {"passed": b["passed"], "run": b["run"]} return {"batches": len(records), "models": models, "cpu_seconds": cpu if cpu_known else None, "first_start": min(starts) if starts else None}
def watts_per_cpu(inv): if not inv: return None if isinstance(inv.get("lane_watts_per_cpu"), (int, float)): return float(inv["lane_watts_per_cpu"]) v = inv.get("verification_tier") or {} machines, cores, watts = v.get("machines"), v.get("total_cores"), v.get("machine_watts") if all(isinstance(x, (int, float)) and x > 0 for x in (machines, cores, watts)): return watts * machines / cores return None
def pool_of(inv, model): """(cards, card watts) serving a model's pool, or None when the inventory does not say.""" if not inv: return None pools = inv.get("model_pools") or {} p = pools.get(model) if not isinstance(p, dict) or not isinstance(p.get("cards"), (int, float)): return None watts = p.get("card_watts") if not isinstance(watts, (int, float)): watts = (inv.get("sampling_tier") or {}).get("card_watts") return p["cards"], watts if isinstance(watts, (int, float)) else None
# ---------------------------------------------------------------- wording
def duration(seconds): s = int(round(seconds)) if s < 60: return f"{s} s" m = s // 60 if m < 60: return f"{m} min" h, m = divmod(m, 60) if h < 24: return f"{h} h {m} min" if m else f"{h} h" d, h = divmod(h, 24) return f"{d} d {h} h" if h else f"{d} d"
def hours(seconds, unit): h = seconds / 3600 return f"{h:.2f} {unit}" if h < 10 else f"{h:.1f} {unit}"
def tokens(n): n = int(n) if n >= 1_000_000: return f"{n / 1_000_000:.1f}M" if n >= 1_000: return f"{n / 1_000:.1f}k" return str(n)
def energy(wh): text = f"{wh / 1000:.2f} kWh" if wh >= 1000 else f"{wh:.0f} Wh" kettles = wh / KETTLE_WH if kettles >= 1.5: return f"~{text}, about {kettles:.0f} kettles of tea" return f"~{text}, less than two kettles of tea"
def clip(text, n): return text if len(text) <= n else text[: n - 3] + "..."
def title_for(features): if len(features) == 1: f = features[0] verb = "now supports" if f["to"] == "complete" else "now partly supports" return clip(f"nev {verb} {f['name']}", LIMITS["title"]) head = f"nev moves {len(features)} web features forward: " names = [f["name"] for f in features] for shown in range(len(names), 0, -1): rest = len(names) - shown t = head + ", ".join(names[:shown]) + (f" and {rest} more" if rest else "") if len(t) <= LIMITS["title"]: return t return clip(head, LIMITS["title"])
def description_for(features): lines = [] for f in features[:MAX_FEATURE_LINES]: was = "new to the map" if f["from"] is None else f"was {f['from']}" lines.append(f"**[{f['name']}](https://webstatus.dev/features/{f['feature']})**: " f"{STATUS_WORDS[f['to']]} ({was})") if len(features) > MAX_FEATURE_LINES: lines.append(f"...and {len(features) - MAX_FEATURE_LINES} more") return clip("\n".join(lines), LIMITS["description"])
# ---------------------------------------------------------------- a job landing
def safe_name(name): return isinstance(name, str) and name != "" and all(c.isalnum() or c in "._-" for c in name)
def spec_title(root, name): """The first `# ` heading of specs/<name>.md, past its front matter; None without one.""" if not safe_name(name): return None try: lines = (Path(root) / "specs" / f"{name}.md").read_text().splitlines() except OSError: return None for line in lines: if line.startswith("# "): return line[2:].strip() or None return None
def admitted(root, result): """One line per provenance file a stage of the job read: the generated function, the model that wrote it, and the cases it passed out of those run against the browsers.""" paths = sorted({p for s in result.get("stages") or [] if isinstance(s, dict) for p in (s.get("artifacts") or {}) if isinstance(p, str) and p.endswith(".provenance.json")}) lines = [] for p in paths: if p.startswith("/") or ".." in Path(p).parts: continue try: prov = json.loads((Path(root) / p).read_text()) except (OSError, ValueError): continue if not isinstance(prov, dict): continue fn = Path(p).name[: -len(".provenance.json")] v = prov.get("verification") if isinstance(prov.get("verification"), dict) else {} run, passed = v.get("cases_run"), v.get("cases_passed") parts = [] if isinstance(prov.get("model"), str): parts.append(f"written by {prov['model']}") if isinstance(run, int) and isinstance(passed, int): parts.append(f"{passed}/{run} cases pass against the three browsers") lines.append(f"admitted the generated function `{fn}`" + (f" ({', '.join(parts)})" if parts else "")) return lines
def run_seconds(result): """From the first round's start to the last round's end (the result is published just after); None when the rounds do not say.""" rounds = [r for r in result.get("rounds") or [] if isinstance(r, dict)] starts = [r["started_at"] for r in rounds if isinstance(r.get("started_at"), (int, float))] ends = [r["ended_at"] for r in rounds if isinstance(r.get("ended_at"), (int, float))] if not starts or not ends: return None return max(0, max(ends) - min(starts))
def unmoved_reason(result, snapshot): """Why the map did not move, as bench/expected-features-check.py would put it.""" expected = [x for x in result.get("expected_features") or [] if isinstance(x, str)] if not expected: return "a check-only job: it declares no web feature" if snapshot is None: return None reasons = [] for fid in expected: entry = snapshot.get(fid) status = entry.get("status") if isinstance(entry, dict) else None if status in RANK: reasons.append(f"{fid} already {status}") elif status is None: reasons.append(f"{fid} absent from the map") else: reasons.append(f"no map rule moves {fid}, which stays {status}") return "; ".join(reasons)
def read_snapshot(path): try: features = json.loads(Path(path).read_text()).get("features") except (OSError, ValueError, AttributeError): return None return features if isinstance(features, dict) else None
def job_parts(result, root, features, snapshot): """Title, description, run time, name and status of a job landing's post.""" name = result.get("name") if isinstance(result.get("name"), str) else "a job" lines = [] heading = spec_title(root, name) if heading: lines.append(heading) lines += admitted(root, result) if features: lines.append(description_for(features)) else: reason = unmoved_reason(result, snapshot) lines.append("map: no status change" + (f" ({reason})" if reason else "")) title = title_for(features) if features else clip(f"nev landed {name}", LIMITS["title"]) return {"name": name, "status": result.get("status") or "unknown", "title": title, "description": clip("\n".join(lines), LIMITS["description"]), "run_seconds": run_seconds(result)}
def build_body(features, track, records, inv, since, now, job=None): """The embed. `job` (from job_parts) makes it a job landing's post: its title, description, run time and footer; otherwise the features alone make the post.""" t = totals(records) if records else None fields = [] footer = [] if job and job["run_seconds"] is not None: fields.append({"name": "Job run", "value": duration(job["run_seconds"]), "inline": True}) if t and t["models"]: ranked = sorted(t["models"].items(), key=lambda kv: (-kv[1]["candidates"], kv[0])) for name, m in ranked[:MAX_MODEL_FIELDS]: best = (f"best {m['best']['passed']}/{m['best']['run']}" if m["best"] else "none scored") value = (f"{m['candidates']:g} drawn, {m['compiled']:g} compiled\n{best}\n" f"{tokens(m['completion_tokens'])} tokens") fields.append({"name": clip(name, LIMITS["field_name"]), "value": value, "inline": True}) if len(ranked) > MAX_MODEL_FIELDS: rest = ranked[MAX_MODEL_FIELDS:] fields.append({"name": f"{len(rest)} more models", "value": f"{sum(m['candidates'] for _, m in rest):g} drawn", "inline": True})
start_points = [x for x in (since, t["first_start"] if t else None) if x is not None] # A job's own run time replaces the wall time: two spans side by side read as a contradiction. if start_points and not (job and job["run_seconds"] is not None): fields.append({"name": "Wall time", "value": duration(max(0, now - min(start_points))), "inline": True})
cpu_wh = gpu_wh = None if t and t["cpu_seconds"] is not None: fields.append({"name": "CPU time", "value": hours(t["cpu_seconds"], "CPU-hours"), "inline": True}) w = watts_per_cpu(inv) if w is not None: cpu_wh = t["cpu_seconds"] / 3600 * w if t and t["models"]: gpu_s, wh, missing = 0.0, 0.0, 0 for name, m in t["models"].items(): pool = pool_of(inv, name) if pool is None: missing += 1 continue cards, watts = pool gpu_s += m["draw_seconds"] * cards if watts is not None: wh += m["draw_seconds"] * cards / 3600 * watts if missing < len(t["models"]): value = hours(gpu_s, "GPU-hours") if missing: value += f" ({missing} model{'s' if missing > 1 else ''} not counted)" fields.append({"name": "GPU time", "value": value, "inline": True}) gpu_wh = wh if cpu_wh is not None or gpu_wh is not None: fields.append({"name": "Energy (estimate)", "value": energy((cpu_wh or 0) + (gpu_wh or 0)), "inline": True})
if job: footer.append(f"job {job['name']}" + (f" ({job['status']})" if job["status"] != "passed" else "")) if track: footer.append(f"track {track}") if t: footer.append(f"{t['batches']} model batch{'es' if t['batches'] != 1 else ''}") else: footer.append("no model batches on record") footer.append("energy from rated watts, not metered")
if job: title, description = job["title"], job["description"] else: title, description = title_for(features), description_for(features) body = {"title": title, "description": description, "fields": fields[: LIMITS["fields"]], "footer": clip(" | ".join(footer), LIMITS["footer"])} for f in body["fields"]: f["value"] = clip(f["value"], LIMITS["field_value"]) over = embed_size(body) - LIMITS["total"] if over > 0: body["description"] = clip(body["description"], max(10, len(body["description"]) - over)) return body
def embed_size(body): return (len(body["title"]) + len(body.get("description", "")) + len(body.get("footer", "")) + sum(len(f["name"]) + len(f["value"]) for f in body.get("fields", [])))
# ---------------------------------------------------------------- posting
def post(body): url, token = setting("NEV_ANNOUNCE_URL"), setting("NEV_ANNOUNCE_TOKEN") if not url or not token: say("skipped: NEV_ANNOUNCE_URL or NEV_ANNOUNCE_TOKEN is not set") return 0 req = urllib.request.Request( url.rstrip("/") + "/announce", data=json.dumps(body).encode(), method="POST", headers={"Content-Type": "application/json", "Authorization": f"Bearer {token}"}) try: with urllib.request.urlopen(req, timeout=20) as r: r.read() except urllib.error.HTTPError as e: say(f"post failed: HTTP {e.code}") return 1 except (urllib.error.URLError, OSError, ValueError) as e: # The reason can name the host, so only its kind is printed. say(f"post failed: could not reach the announcement service ({type(e).__name__})") return 1 say(f"posted {body['title']!r}") return 0
def main(): ap = argparse.ArgumentParser(add_help=True) ap.add_argument("--events", required=True) ap.add_argument("--track", default="") ap.add_argument("--ledger", default="") ap.add_argument("--since", type=float, default=None) ap.add_argument("--now", type=float, default=None) ap.add_argument("--inventory", default="") ap.add_argument("--result", default="") ap.add_argument("--root", default=str(ROOT)) ap.add_argument("--snapshot", default="") ap.add_argument("--dry-run", action="store_true") a = ap.parse_args()
result = None if a.result: try: result = json.loads(Path(a.result).read_text()) if not isinstance(result, dict): raise ValueError("not a JSON object") except (OSError, ValueError) as e: say(f"cannot read the job result ({getattr(e, 'strerror', None) or e}); " "posting only if a feature moved") result = None try: features = read_events(a.events) except OSError as e: if result is None: say(f"not posted: cannot read the events ({e.strerror})") return 1 say(f"cannot read the events ({e.strerror}); the post names no feature") features = [] if not features and result is None: say("no feature reached partial or complete; nothing posted") return 0 try: text = find_ledger(a.track, a.ledger) except OSError as e: say(f"no ledger ({e.strerror})") text = None records = parse_ledger(text) if text else [] if not records and result is not None: records = [r for r in result.get("ledger") or [] if isinstance(r, dict) and r.get("format") == 1] if a.track and not records: say(f"no ledger entries for {a.track}; the post carries no model numbers") job = None if result is not None: snapshot = read_snapshot(a.snapshot or Path(a.root) / "docs" / "web-features-map.json") job = job_parts(result, a.root, features, snapshot) body = build_body(features, a.track, records, read_inventory(a.inventory), a.since, a.now if a.now is not None else time.time(), job) if a.dry_run: print(json.dumps(body, indent=2, ensure_ascii=False)) return 0 return post(body)
if __name__ == "__main__": sys.exit(main())