#!/usr/bin/env python3 """Build and validate the WP44 computer-science ontology. Reads the mapping data modules in taxonomy/src/ (area_*.py, areas.py, unmapped_cs.py), validates them against the OpenAlex topic export and the DBLP stream export, and writes: taxonomy/ontology.yaml (schema of agents/briefs/AGENT_D_TAXONOMY.md) taxonomy/topic_map.csv (one row per (topic_id, category_id, weight)) taxonomy/ONTOLOGY.md (generated section between / markers; hand-written prose outside the markers is preserved) Run: /opt/homebrew/bin/python3 taxonomy/build_ontology.py [--no-write] Exit status is non-zero if any validation fails. """ from __future__ import annotations import csv import datetime as dt import glob import importlib.util import os import re import sys from collections import Counter, defaultdict HERE = os.path.dirname(os.path.abspath(__file__)) SRC = os.path.join(HERE, "src") TOPICS_CSV = os.path.join(HERE, "openalex_topics_2026-10-14.csv") STREAMS_CSV = os.path.join(HERE, "dblp_streams_2018_2026.csv") OUT_YAML = os.path.join(HERE, "ontology.yaml") OUT_MAP = os.path.join(HERE, "topic_map.csv") OUT_MD = os.path.join(HERE, "ONTOLOGY.md") VERSION, CREATED = "1.0", "2026-11-03" CS_FIELD = "17" TOP_STREAMS = 600 MARK_BEGIN, MARK_END = "", "" FIELDS = {"title", "abstract", "both"} LIST_KEYS = ("topics", "subfields", "arxiv", "streams", "keywords", "exclusions", "related", "parent_of", "child_of") errors: list[str] = [] def err(msg: str) -> None: errors.append(msg) def load_module(path: str): name = os.path.splitext(os.path.basename(path))[0] spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) # type: ignore[union-attr] return mod def wt_ok(w) -> bool: return isinstance(w, (int, float)) and 0 < float(w) <= 1 def main(write: bool = True) -> int: topics = {r["topic_id"]: r for r in csv.DictReader(open(TOPICS_CSV, encoding="utf-8"))} subfields = {int(r["subfield_id"]): r["subfield"] for r in topics.values()} streams = {r["stream"]: r for r in csv.DictReader(open(STREAMS_CSV, encoding="utf-8"))} areas_mod = load_module(os.path.join(SRC, "areas.py")) unmapped_mod = load_module(os.path.join(SRC, "unmapped_cs.py")) AREAS = {a[0]: {"id": a[0], "name": a[1], "description": a[2]} for a in areas_mod.AREAS} ARXIV = areas_mod.ARXIV_CATEGORIES cats: list[dict] = [] for path in sorted(glob.glob(os.path.join(SRC, "area_*.py"))): mod = load_module(path) for c in mod.CATEGORIES: c["_module"] = os.path.basename(path) cats.append(c) cats.sort(key=lambda c: c["number"]) ids = [c["id"] for c in cats] idset = set(ids) # ---- structural validation ------------------------------------------------------------------------------------- if len(cats) != 40: err(f"expected 40 categories, found {len(cats)}") if len(idset) != len(ids): err(f"duplicate category ids: {[i for i, n in Counter(ids).items() if n > 1]}") if sorted(c["number"] for c in cats) != list(range(1, 41)): err("category numbers are not exactly 1..40") for c in cats: cid = c["id"] if not re.fullmatch(r"[a-z][a-z0-9_]*", cid): err(f"{cid}: id is not snake_case") if c["area"] not in AREAS: err(f"{cid}: unknown area {c['area']}") if not isinstance(c.get("definition"), str) or len(c["definition"]) < 60: err(f"{cid}: definition missing or too short") if not c.get("provenance"): err(f"{cid}: provenance missing") for k in LIST_KEYS: if not isinstance(c.get(k), list): err(f"{cid}: field {k} must be a list") # duplicates inside signal lists for k, keyfn in (("topics", lambda t: t[0]), ("subfields", lambda t: t[0]), ("arxiv", lambda t: t[0]), ("streams", lambda t: t[0]), ("keywords", lambda t: t[0])): keys = [keyfn(t) for t in c[k]] dup = [x for x, n in Counter(keys).items() if n > 1] if dup: err(f"{cid}: duplicate entries in {k}: {dup}") for tid, w, note in c["topics"]: if tid not in topics: err(f"{cid}: unknown OpenAlex topic id {tid}") if not wt_ok(w): err(f"{cid}: topic {tid} weight {w} not in (0,1]") if not note: err(f"{cid}: topic {tid} has no note") for sid, w in c["subfields"]: if sid not in subfields: err(f"{cid}: unknown subfield id {sid}") if not wt_ok(w): err(f"{cid}: subfield {sid} weight {w} not in (0,1]") for a, w in c["arxiv"]: if a not in ARXIV: err(f"{cid}: arXiv category {a} not in admitted list") if not wt_ok(w): err(f"{cid}: arXiv {a} weight {w} not in (0,1]") for s, w in c["streams"]: if s not in streams: err(f"{cid}: unknown DBLP stream {s}") if not wt_ok(w): err(f"{cid}: stream {s} weight {w} not in (0,1]") for pat, w, field in c["keywords"]: try: re.compile(pat) except re.error as e: err(f"{cid}: keyword regex does not compile: {pat[:60]}... ({e})") if not wt_ok(w): err(f"{cid}: keyword weight {w} not in (0,1]: {pat[:40]}") if field not in FIELDS: err(f"{cid}: keyword field {field!r} invalid") for pat in c["exclusions"]: try: re.compile(pat) except re.error as e: err(f"{cid}: exclusion regex does not compile: {pat[:60]}... ({e})") for k in ("related", "parent_of", "child_of"): for other in c[k]: if other not in idset: err(f"{cid}: {k} references unknown category {other}") if other == cid: err(f"{cid}: {k} references itself") if len(set(c["related"])) != len(c["related"]): err(f"{cid}: duplicate related entries") # parent/child symmetry by_id = {c["id"]: c for c in cats} for c in cats: for ch in c["parent_of"]: if ch in by_id and c["id"] not in by_id[ch]["child_of"]: err(f"{c['id']} parent_of {ch} but {ch} does not declare child_of {c['id']}") for p in c["child_of"]: if p in by_id and c["id"] not in by_id[p]["parent_of"]: err(f"{c['id']} child_of {p} but {p} does not declare parent_of {c['id']}") # keyword sufficiency for the categories the brief singles out must_keyword = {"llm", "foundation_models", "multimodal_ai", "generative_ai", "ai_agents", "ai_safety", "federated_learning", "sustainable_computing", "autonomous_scientific_discovery", "privacy_preserving_computation", "blockchain", "quantum_computing", "edge_computing"} kw_sufficient = {c["id"]: any(float(w) >= 1.0 for _, w, _ in c["keywords"]) for c in cats} for cid in must_keyword: if cid in by_id and not kw_sufficient[cid]: err(f"{cid}: must have at least one keyword pattern of weight 1.0 (threshold reachable alone)") # ---- coverage statistics --------------------------------------------------------------------------------------- topic_cats: dict[str, list[tuple[str, float]]] = defaultdict(list) for c in cats: for tid, w, _ in c["topics"]: topic_cats[tid].append((c["id"], float(w))) cs_topics = {t for t, r in topics.items() if r["field_id"] == CS_FIELD} cs_mapped = sorted(t for t in cs_topics if t in topic_cats) cs_unmapped = sorted(t for t in cs_topics if t not in topic_cats) reasons = unmapped_mod.UNMAPPED_REASONS for t in cs_unmapped: if t not in reasons: err(f"unmapped CS topic {t} ({topics[t]['display_name']}, subfield {topics[t]['subfield_id']}, " f"{topics[t]['works_count']} works) has no reason in src/unmapped_cs.py") for t in reasons: if t not in topics: err(f"unmapped_cs.py: unknown topic id {t}") elif t in topic_cats: err(f"unmapped_cs.py: topic {t} is listed as unmapped but is mapped to {[c for c, _ in topic_cats[t]]}") noncs_mapped = sorted(t for t in topic_cats if topics[t]["field_id"] != CS_FIELD) noncs_by_field = Counter(f"{topics[t]['field_id']} {topics[t]['field']}" for t in noncs_mapped) stream_cats: dict[str, list[tuple[str, float]]] = defaultdict(list) for c in cats: for s, w in c["streams"]: stream_cats[s].append((c["id"], float(w))) ranked = sorted(streams.values(), key=lambda r: -int(r["n"])) top = [r for r in ranked[:TOP_STREAMS] if r["stream"] != "journals/corr"] top_mapped = [r for r in top if r["stream"] in stream_cats] top_unmapped = [r for r in top if r["stream"] not in stream_cats] total_recs = sum(int(r["n"]) for r in ranked if r["stream"] != "journals/corr") mapped_recs = sum(int(streams[s]["n"]) for s in stream_cats if s in streams and s != "journals/corr") if errors: print("VALIDATION FAILED") for e in errors: print(" -", e) return 1 # ---- outputs --------------------------------------------------------------------------------------------------- import yaml # imported late so validation errors are reported even without pyyaml def cat_yaml(c: dict) -> dict: return { "id": c["id"], "number": c["number"], "name": c["name"], "area": c["area"], "definition": c["definition"], "openalex_topics": [{"id": t, "weight": float(w), "note": n} for t, w, n in c["topics"]], "openalex_subfields": [{"id": int(s), "weight": float(w)} for s, w in c["subfields"]], "arxiv_categories": [{"id": a, "weight": float(w)} for a, w in c["arxiv"]], "dblp_streams": [{"key": s, "weight": float(w)} for s, w in c["streams"]], "keywords": [{"pattern": p, "weight": float(w), "field": f} for p, w, f in c["keywords"]], "exclusions": list(c["exclusions"]), "related": list(c["related"]), "parent_of": list(c["parent_of"]), "child_of": list(c["child_of"]), "provenance": c["provenance"], } area_leaves = defaultdict(list) for c in cats: area_leaves[c["area"]].append(c["id"]) doc = { "version": VERSION, "created": CREATED, "sources": {"openalex_topics": os.path.basename(TOPICS_CSV), "dblp_streams": os.path.basename(STREAMS_CSV), "generator": "taxonomy/build_ontology.py", "data_modules": "taxonomy/src/*.py"}, "areas": [{"id": a["id"], "name": a["name"], "description": a["description"], "leaves": area_leaves[a["id"]]} for a in AREAS.values()], "categories": [cat_yaml(c) for c in cats], "assignment_rule": { "score": "sum of weights of matched signals, each signal type capped at 1.0, then sum over signal types", "signal_types": list(areas_mod.SIGNAL_TYPES), "threshold": 1.0, "primary": "argmax of score; ties broken by the more specific category (child before parent), then by lower number", "exclusions": "if any exclusion regex matches the concatenated title+abstract, the category is vetoed for that work", "keyword_field": "title patterns are matched on the title only; abstract on the abstract only; both on title+abstract", "notes": "OpenAlex topic weights assume the work's primary topic (or any of its up-to-three topics; the classifier " "takes the sum of matched topic weights capped at 1.0 before capping). Weights below 1.0 are deliberate: such signals must be " "corroborated by another signal type to reach the threshold.", }, } if write: with open(OUT_YAML, "w", encoding="utf-8") as fh: yaml.safe_dump(doc, fh, allow_unicode=True, sort_keys=False, width=1000) # round-trip check: every regex survives YAML serialisation back = yaml.safe_load(open(OUT_YAML, encoding="utf-8")) for c0, c1 in zip(cats, back["categories"]): for (p, _, _), k in zip(c0["keywords"], c1["keywords"]): assert p == k["pattern"], f"YAML round-trip altered a regex in {c0['id']}" re.compile(k["pattern"]) for p, q in zip(c0["exclusions"], c1["exclusions"]): assert p == q re.compile(q) with open(OUT_MAP, "w", encoding="utf-8", newline="") as fh: w = csv.writer(fh) w.writerow(["topic_id", "category_id", "weight", "topic_name", "field_id", "field", "subfield_id", "subfield", "works_count", "note"]) for c in cats: for tid, wt, note in sorted(c["topics"], key=lambda t: (-float(t[1]), t[0])): r = topics[tid] w.writerow([tid, c["id"], f"{float(wt):.2f}", r["display_name"], r["field_id"], r["field"], r["subfield_id"], r["subfield"], r["works_count"], note]) # ---- generated markdown ---------------------------------------------------------------------------------------- L: list[str] = [] L.append(f"_Generated by `taxonomy/build_ontology.py` on {dt.date.today().isoformat()} from " f"`{os.path.basename(TOPICS_CSV)}` ({len(topics):,} topics) and `{os.path.basename(STREAMS_CSV)}` " f"({len(streams):,} streams). Do not edit between the markers._\n") L.append("## Areas and leaves\n") L.append("| Area | Name | Leaves |") L.append("|---|---|---|") for a in AREAS.values(): L.append(f"| `{a['id']}` | {a['name']} | {', '.join('`'+x+'`' for x in area_leaves[a['id']])} |") L.append("\n## The 40 categories\n") for c in cats: L.append(f"**{c['number']}. {c['name']}** (`{c['id']}`, area `{c['area']}`) ") L.append(f"{c['definition']} ") edges = [] if c["child_of"]: edges.append("child of " + ", ".join(f"`{x}`" for x in c["child_of"])) if c["parent_of"]: edges.append("parent of " + ", ".join(f"`{x}`" for x in c["parent_of"])) if c["related"]: edges.append("related to " + ", ".join(f"`{x}`" for x in c["related"])) L.append("_" + "; ".join(edges) + "._\n") L.append("## Hierarchy (parent → child edges)\n") for c in cats: for ch in c["parent_of"]: L.append(f"- `{c['id']}` → `{ch}`") L.append("\n## Signal counts per category\n") L.append("| # | id | topics | subfields | arXiv | DBLP streams | keywords | exclusions | keyword alone reaches threshold |") L.append("|---|---|---|---|---|---|---|---|---|") for c in cats: L.append(f"| {c['number']} | `{c['id']}` | {len(c['topics'])} | {len(c['subfields'])} | {len(c['arxiv'])} | " f"{len(c['streams'])} | {len(c['keywords'])} | {len(c['exclusions'])} | {'yes' if kw_sufficient[c['id']] else 'no'} |") L.append(f"\nTotals: {sum(len(c['topics']) for c in cats)} topic→category weights over {len(topic_cats)} distinct topics; " f"{sum(len(c['streams']) for c in cats)} stream→category weights over {len(stream_cats)} distinct streams; " f"{sum(len(c['keywords']) for c in cats)} keyword patterns; {sum(len(c['exclusions']) for c in cats)} exclusion patterns.\n") L.append("## OpenAlex coverage\n") L.append(f"- Field 17 (Computer Science): **{len(cs_mapped)} of {len(cs_topics)} topics mapped** " f"({100*len(cs_mapped)/len(cs_topics):.1f} %); {len(cs_unmapped)} left unmapped with an explicit reason below.") L.append(f"- Topics mapped from other fields: {len(noncs_mapped)} — " + "; ".join(f"{k}: {v}" for k, v in sorted(noncs_by_field.items())) + ".") multi = sum(1 for t, l in topic_cats.items() if len(l) > 1) L.append(f"- Topics mapped to more than one category: {multi} of {len(topic_cats)}.") L.append("\n### Mapped CS topics per subfield\n") L.append("| Subfield | topics | mapped | unmapped |") L.append("|---|---|---|---|") bysf = defaultdict(lambda: [0, 0]) for t in cs_topics: bysf[(topics[t]["subfield_id"], topics[t]["subfield"])][0 if t in topic_cats else 1] += 1 for (sid, sname), (m, u) in sorted(bysf.items()): L.append(f"| {sid} {sname} | {m+u} | {m} | {u} |") L.append("\n### Unmapped CS topics (field 17) with reason\n") L.append("| topic | name | subfield | works | reason |") L.append("|---|---|---|---|---|") for t in sorted(cs_unmapped, key=lambda t: -int(topics[t]["works_count"])): r = topics[t] L.append(f"| {t} | {r['display_name']} | {r['subfield_id']} | {int(r['works_count']):,} | {reasons[t]} |") L.append("\n## DBLP coverage\n") L.append(f"- Streams mapped: **{len(stream_cats)}** (any rank); records covered (2018–2026, excluding CoRR): " f"{mapped_recs:,} of {total_recs:,} ({100*mapped_recs/total_recs:.1f} %).") L.append(f"- Of the {len(top)} largest streams (CoRR excluded), {len(top_mapped)} are mapped " f"({100*len(top_mapped)/len(top):.1f} %). The {len(top_unmapped)} unmapped large streams are listed below; they " f"are either out of scope (medicine, geoscience, education, management, electronics) or too generic to be a " f"category signal (IEEE Access, Sensors, Symmetry).") L.append("\n| stream | records | name |") L.append("|---|---|---|") for r in top_unmapped: L.append(f"| `{r['stream']}` | {int(r['n']):,} | {(r['journal'] or r['booktitle'])} |") L.append("\n## Topics mapped to several categories (audit list)\n") L.append("| topic | name | categories (weight) |") L.append("|---|---|---|") for t in sorted(topic_cats, key=lambda t: (-len(topic_cats[t]), t)): if len(topic_cats[t]) > 1: L.append(f"| {t} | {topics[t]['display_name']} | " + ", ".join(f"`{c}` ({w:.1f})" for c, w in sorted(topic_cats[t], key=lambda x: -x[1])) + " |") generated = "\n".join(L) + "\n" if write: if os.path.exists(OUT_MD): text = open(OUT_MD, encoding="utf-8").read() if MARK_BEGIN in text and MARK_END in text: pre = text.split(MARK_BEGIN)[0] post = text.split(MARK_END)[1] text = pre + MARK_BEGIN + "\n" + generated + MARK_END + post else: text = text.rstrip() + "\n\n" + MARK_BEGIN + "\n" + generated + MARK_END + "\n" else: text = ("# WP44 Ontology of Computer-Science Research\n\n(prose to be written)\n\n" + MARK_BEGIN + "\n" + generated + MARK_END + "\n") open(OUT_MD, "w", encoding="utf-8").write(text) # ---- console summary ------------------------------------------------------------------------------------------- print(f"OK: 40 categories, {len(topic_cats)} topics, {len(stream_cats)} streams, " f"{sum(len(c['keywords']) for c in cats)} keyword patterns") print(f"CS topics mapped {len(cs_mapped)}/{len(cs_topics)}; non-CS topics mapped {len(noncs_mapped)}; " f"top-{TOP_STREAMS} streams mapped {len(top_mapped)}/{len(top)}; records covered {100*mapped_recs/total_recs:.1f}%") print("per category (topics/streams/keywords):") for c in cats: print(f" {c['number']:2d} {c['id']:32s} {len(c['topics']):3d} {len(c['streams']):3d} {len(c['keywords']):3d}" f"{'' if kw_sufficient[c['id']] else ' (no 1.0 keyword)'}") return 0 if __name__ == "__main__": sys.exit(main(write="--no-write" not in sys.argv))