Make the two build scripts importable, and read the real prefix lists

scripts/fleet/build_data.py

@@ -4,6 +4,12 @@
 scanning passes. Reads no checkout: its inputs are the JSON those passes wrote
 under _fleet/data, and its output goes back there. See _fleet/README.md for the
 whole pipeline.
+
+Everything below the constants sits behind main(), so VOC can be imported and
+checked against the other two prefix lists without reading or writing any data
+(#97). Before that, importing this module loaded both JSON inputs and rewrote
+fleet_data.json, so the test that pins the three lists to each other had to
+parse VOC out of the source text instead.
 """
 import os
 REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
@@ -13,29 +19,44 @@
 from roots import CITATION, ORDER
 
 S=DATA
-sub=json.load(open(f"{S}/subsets_summary.json")); cen=json.load(open(f"{S}/prefix_census.json"))
 VOC=["CHEBI","NCBITaxon","GO","ENVO","METPO","ARO","UniProt","InterPro","Pfam","RHEA","PDB","PATO","UBERON","FOODON","BTO","GTDB","KEGG","CAS","MIBiG","NPAtlas","PMID","DOI"]
-edges=[]
-for k,v in sub["edges"].items():
-    a,b=k.split("|")
-    if v["n"]==0: continue
-    edges.append({"a":a,"b":b,"n":v["n"],"by":v["by"],"ex":v["ex"]})
-heat={m:{v:cen[m]["prefixes"].get(v,0) for v in VOC} for m in ORDER}
-cells={k.replace("|","--"):n for k,n in sub["cells"].items()}
-
-# Heatmap columns run left to right from the most widely shared vocabulary to
-# the least: first by how many Mechs ground anything in it, then, for the many
-# ties at nine and at one, by the total records citing it across the fleet.
-# Name last so the order is stable when a vocabulary appears in no records.
-def reach(v): return sum(1 for m in ORDER if heat[m][v])
-def records(v): return sum(cells.get(f"{m}--{v}",0) for m in ORDER)
-# CITATION comes from roots.py, the same list build_subsets.py uses to decide
-# which prefixes get no record lists. Those two have to agree: a citation
-# prefix would sort to the far left on Mech count with nothing to break the
-# tie, which is what the pin exists to prevent.
-VOC_ORDER=sorted((v for v in VOC if v not in CITATION),key=lambda v:(-reach(v),-records(v),v))+[v for v in CITATION if v in VOC]
-for v in VOC_ORDER: print(f"  {v:<10} {reach(v)} mechs {records(v):>9,} records")
-
-json.dump({"order":ORDER,"voc":VOC_ORDER,"heat":heat,"cells":cells,"vocab_edges":edges},open(f"{S}/fleet_data.json","w"),separators=(",",":"),ensure_ascii=False)
-print(len(edges),"edges;",os.path.getsize(f"{S}/fleet_data.json"),"bytes")
-for e in sorted(edges,key=lambda e:-e["n"])[:6]: print(e["a"],e["b"],e["n"],e["by"],[x["label"] for x in e["ex"]])
+
+
+def build(sub, cen):
+    """The blob the page embeds, from the two scanning passes' output."""
+    edges=[]
+    for k,v in sub["edges"].items():
+        a,b=k.split("|")
+        if v["n"]==0: continue
+        edges.append({"a":a,"b":b,"n":v["n"],"by":v["by"],"ex":v["ex"]})
+    heat={m:{v:cen[m]["prefixes"].get(v,0) for v in VOC} for m in ORDER}
+    cells={k.replace("|","--"):n for k,n in sub["cells"].items()}
+
+    # Heatmap columns run left to right from the most widely shared vocabulary to
+    # the least: first by how many Mechs ground anything in it, then, for the many
+    # ties at nine and at one, by the total records citing it across the fleet.
+    # Name last so the order is stable when a vocabulary appears in no records.
+    def reach(v): return sum(1 for m in ORDER if heat[m][v])
+    def records(v): return sum(cells.get(f"{m}--{v}",0) for m in ORDER)
+    # CITATION comes from roots.py, the same list build_subsets.py uses to decide
+    # which prefixes get no record lists. Those two have to agree: a citation
+    # prefix would sort to the far left on Mech count with nothing to break the
+    # tie, which is what the pin exists to prevent.
+    VOC_ORDER=sorted((v for v in VOC if v not in CITATION),key=lambda v:(-reach(v),-records(v),v))+[v for v in CITATION if v in VOC]
+    for v in VOC_ORDER: print(f"  {v:<10} {reach(v)} mechs {records(v):>9,} records")
+
+    return {"order":ORDER,"voc":VOC_ORDER,"heat":heat,"cells":cells,"vocab_edges":edges}
+
+
+def main():
+    sub=json.load(open(f"{S}/subsets_summary.json")); cen=json.load(open(f"{S}/prefix_census.json"))
+    document=build(sub, cen)
+    with open(f"{S}/fleet_data.json","w") as handle:
+        json.dump(document,handle,separators=(",",":"),ensure_ascii=False)
+    edges=document["vocab_edges"]
+    print(len(edges),"edges;",os.path.getsize(f"{S}/fleet_data.json"),"bytes")
+    for e in sorted(edges,key=lambda e:-e["n"])[:6]: print(e["a"],e["b"],e["n"],e["by"],[x["label"] for x in e["ex"]])
+
+
+if __name__ == "__main__":
+    main()

scripts/fleet/build_subsets.py

@@ -33,7 +33,10 @@
  "MediaIngredientMech": GH+"MediaIngredientMech/blob/main/data/ingredients/",
  "CultureMech": GH+"CultureMech/blob/main/data/merge_yaml/merged/",
 }
-MECHS={name: dict(root=mech_root(name), base=SITE_BASE[name]) for name in ORDER}
+# Filled by prepare(). mech_root() touches the filesystem and exits on a
+# missing checkout, so resolving these at import made the module unimportable
+# and its prefix list unreadable without a scan (#97).
+MECHS={}
 PREF=["CHEBI","NCBITaxon","GO","ENVO","METPO","ARO","UniProt","InterPro","Pfam","PATO","UBERON","FOODON","KEGG","CAS","RHEA","PDB","BTO","GTDB","MIBiG","NPAtlas","DOI"]
 NORM={"mibig":"MIBiG","npatlas":"NPAtlas","UniProtKB":"UniProt","PFAM":"Pfam","IPR":"InterPro","cas":"CAS","doi":"DOI","MeSH":"MESH"}
 rx=re.compile(r"\b(CHEBI|NCBITaxon|GO|ENVO|METPO|ARO|UniProtKB|UniProt|InterPro|IPR|Pfam|PFAM|PATO|UBERON|FOODON|KEGG|CAS|cas|RHEA|PDB|BTO|GTDB|mibig|MIBiG|npatlas|NPAtlas|DOI|doi):([A-Za-z0-9_.\-/()]+)")
@@ -46,14 +49,25 @@ def unq(v):
     while len(v)>=2 and v[0]==v[-1] and v[0] in "'\"": v=v[1:-1].strip()
     return v.replace("''","'")
 # HabitatMech publishes one page per record; the slug is matched against these.
-hab_pages=set(os.path.basename(f)[:-5] for f in glob.glob(os.path.join(mech_root("HabitatMech"),"pages","habitats","*.html")))
-if not hab_pages:
-    raise SystemExit("HabitatMech: no pages under pages/habitats; record links would silently be dropped")
+# Also filled by prepare(), for the same reason as MECHS.
+hab_pages=set()
 hab_by_suffix=collections.defaultdict(list)
-for n in hab_pages:
-    parts=n.split("-")
-    for k in range(1,len(parts)): hab_by_suffix["-".join(parts[k:])].append(n)
 hab_collisions=[]
+
+
+def prepare():
+    """Resolve the checkouts and index HabitatMech's pages.
+
+    Everything here reads the filesystem, which is why it is not at import.
+    """
+    MECHS.update({name: dict(root=mech_root(name), base=SITE_BASE[name]) for name in ORDER})
+    hab_pages.update(os.path.basename(f)[:-5]
+                     for f in glob.glob(os.path.join(mech_root("HabitatMech"),"pages","habitats","*.html")))
+    if not hab_pages:
+        raise SystemExit("HabitatMech: no pages under pages/habitats; record links would silently be dropped")
+    for n in hab_pages:
+        parts=n.split("-")
+        for k in range(1,len(parts)): hab_by_suffix["-".join(parts[k:])].append(n)
 def simple_slug(text):
     return re.sub(r"-+","-",re.sub(r"[^a-z0-9]+","-",text.lower())).strip("-")
 def slug_for(m, f, doc_id, doc_label=""):
@@ -131,53 +145,59 @@ def scan(m, keep=None, cap_cell=300):
                                 break
     print(m,"files",nfiles,"unlinked",nolink,"terms",len(terms),flush=True)
     return dict(terms=terms,cells=cells,votes=votes)
-idx={}
-for m in ORDER:
-    if m=="ProteinTraitsMech": continue
-    idx[m]=scan(m)
-union=set().union(*[set(idx[m]["terms"]) for m in idx])
-idx["ProteinTraitsMech"]=scan("ProteinTraitsMech",keep=union)
-# labels
-labels={}
-allterms=set().union(*[set(idx[m]["terms"]) for m in ORDER])
-for t in allterms:
-    p=t.split(":")[0]
-    c=strict.get(t)
-    if c:
-        l,n=c.most_common(1)[0]
-        if n/sum(c.values())>=0.6 and l.count("'")%2==0: labels[t]=l; continue
-    for a in AUTH.get(p,[]):
-        c=idx[a]["votes"].get(t)
+def main():
+    prepare()
+    idx={}
+    for m in ORDER:
+        if m=="ProteinTraitsMech": continue
+        idx[m]=scan(m)
+    union=set().union(*[set(idx[m]["terms"]) for m in idx])
+    idx["ProteinTraitsMech"]=scan("ProteinTraitsMech",keep=union)
+    # labels
+    labels={}
+    allterms=set().union(*[set(idx[m]["terms"]) for m in ORDER])
+    for t in allterms:
+        p=t.split(":")[0]
+        c=strict.get(t)
         if c:
             l,n=c.most_common(1)[0]
-            if n/sum(c.values())>=0.6 and l.count("'")%2==0: labels[t]=l; break
-os.makedirs(f"{OUT}/edges",exist_ok=True); os.makedirs(f"{OUT}/cells",exist_ok=True)
-summary={"edges":{},"cells":{}}
-for a,b in itertools.combinations(ORDER,2):
-    shared=set(idx[a]["terms"])&set(idx[b]["terms"])
-    # An edge counts shared *concepts*, not shared bibliography. roots.CITATION
-    # says why: every Mech cites papers, so counting those "would say only
-    # that". build_data.py already keeps them out of the heatmap ordering and
-    # the cell indexes below already skip them; the edge weight was the one
-    # place that still counted them, because this line read
-    # `!="DOI" or True` and the `or True` made it a no-op (#62).
-    shared={t for t in shared if t.split(":")[0] not in CITATION}
-    if not shared: continue
-    rows=[]
-    for t in shared:
-        ra=idx[a]["terms"][t]; rb=idx[b]["terms"][t]
-        rows.append({"id":t,"l":labels.get(t,""),"na":len(ra),"nb":len(rb),"a":ra[:6],"b":rb[:6]})
-    rows.sort(key=lambda r:(-(min(r["na"],r["nb"])),-(r["na"]+r["nb"]),r["id"]))
-    byp=collections.Counter(t.split(":")[0] for t in shared)
-    doc={"a":a,"b":b,"base":{a:MECHS[a]["base"],b:MECHS[b]["base"]},"n":len(shared),"by":dict(byp.most_common()),"terms":rows}
-    fn=f"{a}--{b}.json"; json.dump(doc,open(f"{OUT}/edges/{fn}","w"),separators=(",",":"),ensure_ascii=False)
-    summary["edges"][f"{a}|{b}"]={"n":len(shared),"by":dict(byp.most_common()),"ex":[{"id":r["id"],"label":r["l"]} for r in rows if r["l"] and r["id"].split(":")[0] not in CITATION][:3]}
-    print("edge",a,b,len(shared),os.path.getsize(f"{OUT}/edges/{fn}")//1024,"KB")
-for m in ORDER:
-    for p,(n,refs) in idx[m]["cells"].items():
-        if p in CITATION: continue
-        fn=f"{m}--{p}.json"
-        json.dump({"mech":m,"prefix":p,"base":MECHS[m]["base"],"total":n,"records":refs},open(f"{OUT}/cells/{fn}","w"),separators=(",",":"),ensure_ascii=False)
-        summary["cells"][f"{m}|{p}"]=n
-json.dump(summary,open(DATA+"/subsets_summary.json","w"),indent=1)
-print("labels resolved",len(labels),"of",len(allterms)); print("habitat pages unresolved (no link emitted):",len(hab_collisions), hab_collisions[:3]); print("done; total size KB:", sum(os.path.getsize(f) for f in glob.glob(f"{OUT}/**/*.json",recursive=True))//1024)
+            if n/sum(c.values())>=0.6 and l.count("'")%2==0: labels[t]=l; continue
+        for a in AUTH.get(p,[]):
+            c=idx[a]["votes"].get(t)
+            if c:
+                l,n=c.most_common(1)[0]
+                if n/sum(c.values())>=0.6 and l.count("'")%2==0: labels[t]=l; break
+    os.makedirs(f"{OUT}/edges",exist_ok=True); os.makedirs(f"{OUT}/cells",exist_ok=True)
+    summary={"edges":{},"cells":{}}
+    for a,b in itertools.combinations(ORDER,2):
+        shared=set(idx[a]["terms"])&set(idx[b]["terms"])
+        # An edge counts shared *concepts*, not shared bibliography. roots.CITATION
+        # says why: every Mech cites papers, so counting those "would say only
+        # that". build_data.py already keeps them out of the heatmap ordering and
+        # the cell indexes below already skip them; the edge weight was the one
+        # place that still counted them, because this line read
+        # `!="DOI" or True` and the `or True` made it a no-op (#62).
+        shared={t for t in shared if t.split(":")[0] not in CITATION}
+        if not shared: continue
+        rows=[]
+        for t in shared:
+            ra=idx[a]["terms"][t]; rb=idx[b]["terms"][t]
+            rows.append({"id":t,"l":labels.get(t,""),"na":len(ra),"nb":len(rb),"a":ra[:6],"b":rb[:6]})
+        rows.sort(key=lambda r:(-(min(r["na"],r["nb"])),-(r["na"]+r["nb"]),r["id"]))
+        byp=collections.Counter(t.split(":")[0] for t in shared)
+        doc={"a":a,"b":b,"base":{a:MECHS[a]["base"],b:MECHS[b]["base"]},"n":len(shared),"by":dict(byp.most_common()),"terms":rows}
+        fn=f"{a}--{b}.json"; json.dump(doc,open(f"{OUT}/edges/{fn}","w"),separators=(",",":"),ensure_ascii=False)
+        summary["edges"][f"{a}|{b}"]={"n":len(shared),"by":dict(byp.most_common()),"ex":[{"id":r["id"],"label":r["l"]} for r in rows if r["l"] and r["id"].split(":")[0] not in CITATION][:3]}
+        print("edge",a,b,len(shared),os.path.getsize(f"{OUT}/edges/{fn}")//1024,"KB")
+    for m in ORDER:
+        for p,(n,refs) in idx[m]["cells"].items():
+            if p in CITATION: continue
+            fn=f"{m}--{p}.json"
+            json.dump({"mech":m,"prefix":p,"base":MECHS[m]["base"],"total":n,"records":refs},open(f"{OUT}/cells/{fn}","w"),separators=(",",":"),ensure_ascii=False)
+            summary["cells"][f"{m}|{p}"]=n
+    json.dump(summary,open(DATA+"/subsets_summary.json","w"),indent=1)
+    print("labels resolved",len(labels),"of",len(allterms)); print("habitat pages unresolved (no link emitted):",len(hab_collisions), hab_collisions[:3]); print("done; total size KB:", sum(os.path.getsize(f) for f in glob.glob(f"{OUT}/**/*.json",recursive=True))//1024)
+
+
+if __name__ == "__main__":
+    main()

tests/test_fleet_page.py

@@ -1,6 +1,5 @@
 """Regression coverage for fleet admission, capability drift and generated output."""
 from copy import deepcopy
-import ast
 import contextlib
 import json
 import os
@@ -278,12 +277,6 @@ def census_sandbox():
                                PYTHONPATH=str(root / "scripts/fleet"),
                                MECHS_ROOT=str(root / "empty"))
 
-# Methods that change a list in place. A read like VOC.index(v) must not be
-# mistaken for one.
-LIST_MUTATORS = {"append", "extend", "insert", "remove", "pop", "clear",
-                 "sort", "reverse", "__setitem__", "__delitem__"}
-
-
 class PrefixListTests(unittest.TestCase):
     """The pipeline carries three hand-maintained prefix lists that must agree.
 
@@ -313,8 +306,15 @@ def literal_prefix(p):
         norm = constants["norm"]
         self.census = {norm.get(literal_prefix(p), literal_prefix(p))
                        for p in constants["P"].split("|")}
-        self.voc = self.literal("build_data.py", "VOC")
-        self.pref = self.literal("build_subsets.py", "PREF")
+        # Imported, not parsed out of the source. Until #97 both modules did
+        # their work at import — build_subsets resolved every checkout and
+        # build_data read and rewrote fleet_data.json — so the lists had to be
+        # recovered from the source text with ast. That could not see a list
+        # rebuilt after its literal, which is what #99 and #101 were about.
+        # Reading the objects the pipeline actually uses retires the whole class.
+        import build_data, build_subsets
+        self.voc = build_data.VOC
+        self.pref = build_subsets.PREF
 
     @staticmethod
     def module_constants():
@@ -330,56 +330,6 @@ def module_constants():
                 f"(#95):\n{done.stderr}")
         return json.loads(done.stdout)
 
-    def literal(self, script, name):
-        """Read a list literal without importing — both scripts scan on import.
-
-        Parsed with ast rather than matched with a regex. A regex sees only the
-        text it matched, so `VOC = VOC + ["BOGUS"]` on the following line would
-        leave the assertion reading a literal the module no longer uses, and a
-        length check cannot notice. Requiring exactly one module-level binding
-        whose value is a plain literal rules both out.
-        """
-        source = (ROOT / "scripts/fleet" / script).read_text()
-        tree = ast.parse(source)
-        # ast.Assign alone misses `VOC += [...]`, `VOC.append(...)`, `VOC[0] = ...`
-        # and a rebinding nested in an `if` — each leaves this reading a list the
-        # module no longer uses (#99). Walking the tree covers those four.
-        # It does NOT cover tuple-unpack rebinding, `del VOC[0]`, `VOC[0] += x`
-        # or a walrus; those are filed rather than chased (#101).
-        #
-        # Only mutating methods count. Flagging every attribute call would fail
-        # on `VOC.index(v)` — a read, and an idiomatic one in a file that
-        # already sorts VOC — with a message asserting a mutation that never
-        # happened, which is a worse trap than the hole it closes.
-        bindings, mutations = [], []
-        for node in ast.walk(tree):
-            if isinstance(node, ast.Assign):
-                for target in node.targets:
-                    if getattr(target, "id", None) == name:
-                        bindings.append(node)
-                    elif (isinstance(target, ast.Subscript)
-                          and getattr(target.value, "id", None) == name):
-                        mutations.append("subscript assignment")
-            elif isinstance(node, (ast.AugAssign, ast.AnnAssign)):
-                if getattr(node.target, "id", None) == name:
-                    bindings.append(node)
-            elif (isinstance(node, ast.Call)
-                  and isinstance(node.func, ast.Attribute)
-                  and getattr(node.func.value, "id", None) == name
-                  and node.func.attr in LIST_MUTATORS):
-                mutations.append(f"{name}.{node.func.attr}()")
-        self.assertEqual(mutations, [], f"{name} is mutated in {script} after it is bound: "
-                                        f"{mutations}; this test reads a single literal")
-        self.assertEqual(len(bindings), 1,
-                         f"{name} is bound or shadowed {len(bindings)} times in {script} "
-                         f"(lines {[node.lineno for node in bindings]}); "
-                         "this test reads a single literal binding")
-        # literal_eval refuses anything that is not a literal, so a computed
-        # value fails here rather than being silently half-read.
-        value = ast.literal_eval(bindings[0].value)
-        self.assertGreater(len(value), 10, f"{name} parsed as {value!r}, which looks wrong")
-        return value
-
     def test_every_heatmap_column_is_a_vocabulary_the_census_counts(self):
         # A column the census never counts renders as a stripe of zeros.
         self.assertEqual([v for v in self.voc if v not in self.census], [])