"""The reading of a window written by an LLM, for the Findings tab of the PanGramGraph page (serve_live.py /reading). The page writes a rule-based reading of the window shown (autoReading in page_template.html). Here an LLM writes it in prose, by a skill document (reading_skill.md: what the numbers mean, a procedure, the hand-written reading of the dnaA window as its reference), in the same five sections for every window, and every sentence names the facts lines it is written from: each one is checked against those lines before the page shows it, and dropped if it does not hold. the browser sends only the window's id: dnaA, or the anchor family of a region window (pgb_region.build; a backbone family). The server builds the facts itself from its own data (facts(D)): the columns, their stretches (spine, fork, variable, the same rules as the page's autoReading), the stretch most and least in regions of plasticity, the genes and their products, the model's accuracy and hesitation, all as display-ready numbers: 2,000-2,900 tokens, after a system message of ~2,400 (the skill, ~1,800, included). No text of the client reaches the LLM. skill: reading_skill.md beside this file (READING_SKILL for another path), read once and put in the system message: what the pangenome words mean (persistent, shell, cloud, regions of plasticity, spots, spine, fork, branch), what the model's three numbers mean and do not mean (named, bits, perplexity), what each section says, the procedure to follow in order (the gene groups and the one or two interpretations included), and two reference readings in the output form: the hand-written one of the dnaA window and one of a window nearly all spine. Its sha is part of the cache key, so a changed skill makes new readings; without the file the prompt keeps the hand-written reading alone. form: the same five paragraphs in the same order for every window, each opening with its tag and a short bold sentence: @start @variable @model @end @whole (SECTIONS), 150 to 200 words asked (WORDS_TARGET), at most 220 served (WORDS_MAX), temperature 0. citations: every facts line carries its id ([W], [G], [C], [M], [S3], [S3.c12], [S3.b40], [Q]) and every sentence ends with the ids of the lines it is written from, in braces. The check is per sentence, against those lines only (stricter than a check over the whole facts): every number must be one they display, in the right role (a share, a share in a region of plasticity and a share named are not exchangeable), and bound to what it is said of: a gene's share to that gene, a column's to that column, a range to the lowest and highest of the columns named (or a range a line shows), "from A to B" to the first and last column, "most hesitant" / "fewest named" to the column and value the cited line states, [G]'s and [C]'s numbers to their own places. Every gene name, identifier and column must stand in them and sit where the sentence puts it; the kind of a column (spine, fork, variable) and the partition of a family as the facts give them; a branch spoken of with its own families and share. Every word inside [[ ]] must come from the products and gene names of the lines cited and the label must fit most of the families at the columns it names (in one operon, two of them); outside [[ ]] no product word. [M] and [C] in their direction; no "readily", no unstated trend ("fewer and fewer"), no "then" pointing back, no "lack", no unstated superlative. A sentence whose facts are not in its citations is dropped, a sentence repeating an earlier one is left out, and the model is asked once again for the whole reading with the offending sentences named (two dropped, a paragraph without its opening sentence, or no interpretation where [Q] offers one); if too little is left (a missing @start or @whole, two missing sections, under 110 words) the page keeps its rule-based reading (status "fallback"). [Q], the groups: the server finds where an interpretation may stand (genes sharing a prefix, a word shared along a run of products, a branch whose families agree) and the words their products share; the model words the label. tidy(text) puts the model's slips of form right without a new call: "92-99%" -> "92 to 99%", "RGP 5%" -> "5% in a region of plasticity", "share 90%" -> "in 90% of chromosomes", a row of more than 6 genes -> `first` to `last`. length: over 220 words, whole sentences are dropped from the least telling sections, then a whole section; never a part of a sentence. The check verifies the numbers, the names and what the words inside [[ ]] rest on, not the prose. cache: per window, in memory and in /.json, also read from pgb/readings/ (the readings the Space carries), keyed by a hash of the facts, the prompt and the model: new data or a new prompt make a new reading. Limits on new generations: per client (X-Forwarded-For only behind a trusted proxy), per client per day, per day for the whole server (count and dollars), one generation per window at a time, two at once (others wait their turn in a short queue), only for requests from the page's own site (Sec-Fetch-Site / Origin). Model: deepseek-ai/DeepSeek-V4.1-Flash through Hugging Face Inference Providers (router.huggingface.co, OpenAI compatible), no reasoning (chat_template_kwargs thinking=false), at most 800 tokens out (median 410 per attempt over the 112 readings of prompt r9, 460 at most). Token: HF_TOKEN (the Space's secret), else the token of `huggingface-cli login`; used only in the request's header (never followed through a redirect), never logged, never written, never sent to the page; the router's error text is not passed on. READING=off disables new readings (cached ones are still served; /health says so) READING_MODEL, READING_PROVIDERS=deepinfra,novita the model, and the providers tried in that order READING_SKILL=reading_skill.md the skill document; READING_TEMPERATURE=0; READING_WORDS=220 the reading's length READING_PER_IP=6 READING_PER_IP_S=600 READING_PER_IP_DAY=30 new readings per client per 600 s, and per day READING_DAILY_CAP=150 new readings per UTC day, the whole server; READING_DAILY_USD=0.30 and dollars per day (every call counted, a call without usage or cut by a timeout at its worst case) READING_TIMEOUT=22 seconds per LLM call (then the next provider); READING_QUEUE=6 generations waiting READING_CACHE=pgb/readings where new readings and the day's counter are written (e.g. /data/readings on a Space with persistent storage); pgb/readings/ is read too TRUST_PROXY=1 read the client from X-Forwarded-For (always on a Space: SPACE_ID) python3 pg_reading.py facts dnaA | the facts of a window (a region: pgb_region.build output) python3 pg_reading.py check dnaA | the checker on a text with its tags and citations, sentence by sentence, against that window's facts, and the reading it leaves python3 pg_reading.py skill the skill document read, its sha and its size python3 pg_reading.py pregen [--chain] [--dry] [dnaA] [anchor ...] write readings ahead (--chain: the windows of the page's Next and Previous buttons from the dnaA window on), into READING_CACHE, without the server """ import os, re, json, time, html, hashlib, threading, collections, ipaddress, urllib.request, urllib.error, urllib.parse, datetime from decimal import Decimal, ROUND_HALF_UP, ROUND_HALF_EVEN, InvalidOperation def _env(k, d, f=float): try: return f(os.environ.get(k, d)) except ValueError: return f(d) MODEL = os.environ.get("READING_MODEL", "deepseek-ai/DeepSeek-V4.1-Flash") PROVIDERS = [p.strip() for p in os.environ.get("READING_PROVIDERS", "deepinfra,novita").split(",") if p.strip()] PRICES = {"deepinfra": (0.2, 0.6), "novita": (0.3, 1.2), "baseten": (0.3, 1.2)} # $ per million tokens in, out (router, 2026-09) URL = "https://router.huggingface.co/v1/chat/completions" MAX_TOKENS = 800 # median 540 out; 700 cut 2 of 130 texts TEMPERATURE = _env("READING_TEMPERATURE", "0") # 0: the same facts and the same skill give the same reading TIMEOUT = _env("READING_TIMEOUT", "22") BUDGET = 70.0 # seconds for a whole generation (two attempts, a provider fallback) PER_IP = _env("READING_PER_IP", "6", int) PER_IP_S = _env("READING_PER_IP_S", "600") PER_IP_DAY = _env("READING_PER_IP_DAY", "30", int) DAILY_CAP = _env("READING_DAILY_CAP", "150", int) DAILY_USD = _env("READING_DAILY_USD", "0.30") BUNDLED = "pgb/readings" # the readings shipped with the code (the Space serves them without a call) CACHE_DIR = os.environ.get("READING_CACHE") or BUNDLED FAIL_TTL = 6 * 3600 # a window whose text failed the checker twice is tried again after this long AUTH_RETRY = 600 # a token refused by the router: tried again after this long MAX_PARALLEL = 2 # LLM generations at once QUEUE_MAX = _env("READING_QUEUE", "6", int) # new generations waiting for one of those slots SLOT_WAIT = 30.0 # seconds a new generation waits for a slot WAITERS = 4 # requests waiting for the generation of the same window PROMPT_VERSION = "r9" # r9: the skill v2, [M] by kind of column, [Q] the groups, numbers bound to their genes and columns WORDS_MAX = _env("READING_WORDS", "220", int) # the length served: whole sentences are dropped to fit under it WORDS_TARGET = _env("READING_TARGET", "200", int) # the length asked of the model: below the limit, so nothing has to be cut WORDS_MIN_ASK = 150 # the shortest reading asked for WORDS_MIN = 110 # what must be left after the drops for the reading to be served SECTIONS = ("start", "variable", "model", "end", "whole") # the sections, in this order, of every reading KEEP = ("start", "whole") # the sections a reading cannot lose DROP_ORDER = ("variable", "model", "start", "end", "whole") # whose last sentence goes first when the text is too long CITES_MAX = 8 # facts lines one sentence may cite (a stretch and the columns it walks) SKILL_PATH = os.environ.get("READING_SKILL") or os.path.join(os.path.dirname(os.path.abspath(__file__)), "reading_skill.md") DNAA = "dnaA" # the id of the page's home window (pgb/graph_pgb.json, all 2,002 genomes) SPACE = bool(os.environ.get("SPACE_ID")) TRUST_PROXY = SPACE or os.environ.get("TRUST_PROXY") == "1" class ReadingError(Exception): """an answer other than a reading: HTTP status, a machine-readable code (the page words it), retry_s""" status, code = 500, "error" def __init__(self, msg, retry_s=None, code=None): super().__init__(msg); self.retry_s = retry_s; self.code = code or type(self).code def body(self): b = dict(error=str(self), code=self.code) if self.retry_s: b["retry_s"] = int(self.retry_s) if isinstance(self, Unavailable): b["available"] = False return b class Unavailable(ReadingError): status, code = 503, "unavailable" # off, no token, token refused, no credit, the day's cap class Limited(ReadingError): status, code = 429, "limit" # this client's new readings class Busy(ReadingError): status, code = 503, "busy" # other readings are being written: retry_s class Upstream(ReadingError): status, code = 502, "upstream" # the LLM could not be reached class Refused(ReadingError): status, code = 403, "cross_site" # a new reading asked for by another site's page # ---------------------------------------------------------------- numbers as the facts display them def D_(x): return Decimal(str(x)) def fmt_pct(v): """a share (0-1) as a percent: whole numbers, one decimal when that would show 0% or 100% for a share that is not""" x = D_(v) * 100 q = x.quantize(Decimal("1"), ROUND_HALF_UP) if (q == 100 and x < 100) or (q == 0 and x > 0): q = x.quantize(Decimal("0.1"), ROUND_HALF_UP) if q == 100 and x < 100: q = Decimal("99.9") if q == 0 and x > 0: q = Decimal("0.1") return f"{q}%" def fmt_bits(v): return f"{D_(v).quantize(Decimal('0.01'), ROUND_HALF_UP)}" def fmt_fam(v): return f"{D_(v).quantize(Decimal('0.1'), ROUND_HALF_UP)}" def fmt_int(v): return f"{int(v):,}" class Facts: """the facts text of a window, every number it displays by kind (pct, bits, fam, int), and for the checker: each line's id and columns and stretch, the columns of each gene, the columns stated most hesitant and fewest named, the stretches most and least in regions of plasticity, the inside/outside comparison. Each line is shown with its id ([W], [S3], [S3.c12]): a sentence of the reading cites the ids of the lines it is written from.""" def __init__(self): self.ids = [] self.lines = []; self.info = []; self.nums = collections.defaultdict(set); self.genes = set(); self.frags = set() self.at = collections.defaultdict(set); self.atp = collections.defaultdict(set); self.hes_cols = set(); self.low_cols = set(); self.plastic = {} self.comp = None; self.unit = "genomes"; self.label = "" self.pend = []; self.pshow = set(); self.pwords = set(); self.pnames = set(); self.kind = None; self.hes_spine = None self.gshare = collections.defaultdict(set); self.pshare = collections.defaultdict(set); self.gtop = None; self.gbits = None def reg(self, s, kind): v = Decimal(s.rstrip("%").replace(",", "")); self.nums[kind].add(v); return s def pct(self, v): return self.reg(fmt_pct(v), "pct") def bits(self, v): return self.reg(fmt_bits(v), "bits") def fam(self, v): return self.reg(fmt_fam(v), "fam") def int(self, v): return self.reg(fmt_int(v), "int") def fmly(self, n, k=None, alt=False): """a family of the line being built: its gene name and the words of its whole product, truncated or not shown or not — the scope of an interpretation is judged over the families of a line, each with what it really is (alt: a second family at that column, not what the stretch is made of)""" self.pend.append(dict(label=n.get("label") or "", col=k, words=pwords(n), alt=alt, part=PART.get(n.get("partition")))) self.pwords |= pwords(n, gene=False) return n def shown(self, s): """a product string the facts show: its words in full, so a product cut short ("…methy…") still carries them""" self.pshow |= {w for w in re.findall(r"[A-Za-z][A-Za-z'-]{2,}", s or "")} return s def name(self, n, short=40, k=None, alt=False): """a family as the facts name it: its gene name, else its product in quotes (k: its column, for the checker)""" self.fmly(n, k, alt) if n.get("named"): self.genes.add(n["label"]) if k is not None: self.at[n["label"]].add(k) return n["label"] lab = n.get("product") or n.get("label") or "unnamed family" if re.fullmatch(r"[A-Z0-9]+_RS\d+", lab): lab = "unnamed family" self.pnames |= {w.lower() for w in re.findall(r"[A-Za-z][A-Za-z'-]{3,}", lab)} # a family the facts name by its product: if k is not None: self.atp[lab.lower()].add(k); self.atp[self.text(lab, short).lower()].add(k) return '"' + self.text(lab, short) + '"' def text(self, s, short=48): """a product: shortened; its words followed by a number ("family 4") are names, not numbers, in a text""" self.shown(s) # its words in full: the cut is only in the display s = s if len(s) <= short else s[:short - 1].rstrip() + "…" for m in re.finditer(r"([A-Za-z][\w'-]*)\s+(\d+)\b(?!\.\d)", s): self.frags.add(f"{m.group(1)} {m.group(2)}".lower()) return s def add(self, s, cols=(), head=None, glob=False, id=None): """a line; id: the id the reading cites it by; cols: the columns it is about; head: the index of its stretch's line; glob: about the whole window""" self.ids.append(id or f"F{len(self.lines)}") fams, show = self.pend, self.pshow; self.pend = []; self.pshow = set() self.lines.append(s); self.info.append(dict(cols=set(cols), head=head, glob=glob, fams=fams, show=show)) return len(self.lines) - 1 def txt(self): return "\n".join(f"[{i}] {s}" for i, s in zip(self.ids, self.lines)) def index(self): return {i: k for k, i in enumerate(self.ids)} def columns(D): """the window's columns as the page's autoReading sees them: the most carried family (at its most common column), its share of the genomes, the share of the second one; spine >= 90 %, fork: a second one >= 10 %""" W, NG = D["meta"]["window"], D["meta"]["n_genomes"] pl = {r["locus"]: r for r in D["per_locus"]}; by = collections.defaultdict(list) for n in D["nodes"]: by[n["locus"]].append(n) cols = [] for k in range(W): here = sorted(by.get(k, []), key=lambda n: -n["n"]) if not here: continue top = here[0]; tot = sum(n["n"] for n in here) or 1 s1 = top["n"] / NG; s2 = here[1]["n"] / NG if len(here) > 1 else 0 r = pl.get(k) rgp = r["rgp"] if r and r.get("rgp") is not None else sum(n["n"] * n["rgp"] for n in here) / tot cols.append(dict(k=k, top=top, here=here, s1=s1, kind="spine" if s1 >= 0.9 else "fork" if s2 >= 0.1 else "variable", alts=[n for n in here[1:] if n["n"] / NG >= 0.1], rgp=rgp, r=r)) runs = [] for c in cols: if runs and runs[-1]["kind"] == c["kind"] and c["k"] == runs[-1]["b"] + 1: runs[-1]["b"] = c["k"]; runs[-1]["cols"].append(c) else: runs.append(dict(kind=c["kind"], a=c["k"], b=c["k"], cols=[c])) return cols, runs def pwords(n, gene=True): """the words of a family: its whole product and its gene name, split on hyphens, slashes and apostrophes too ("pseudouridine-5'-phosphate glycosidase" gives phosphate and glycosidase), lowercase, the generic ones dropped""" out = {w.lower() for w in re.findall(r"[A-Za-z][A-Za-z]{2,}", n.get("product") or "")} out |= {w.lower() for w in re.findall(r"\b[A-Za-z0-9]+(?:-[A-Za-z0-9]+)+\b", n.get("product") or "") if re.search(r"[A-Za-z]", w)} if gene and n.get("named") and n.get("label"): out.add(n["label"].lower()) return out - GENERIC PART = {"persistent": "P", "shell": "S", "cloud": "C"} GENERIC = set("""protein family domain domain-containing containing putative probable transcriptional regulator subunit hypothetical uncharacterized dna-binding system component type like bifunctional membrane inner outer protein-containing lipoprotein periplasmic cytoplasmic small large chain""".split()) def words(p): """what a product says: its words that are neither generic nor a protein name""" return {w.lower() for w in re.findall(r"[A-Za-z][\w-]{3,}", p or "") if w.lower() not in GENERIC and not re.fullmatch(r"[A-Z][a-z]{2}[A-Z]\w*", w)} def informative(p): """a product that says more than its gene name: a long word, or one with capitals or digits""" if not p or p.lower() == "hypothetical protein": return False ws = [w.lower() for w in re.findall(r"[A-Za-z][\w/-]{3,}", p)] return any((w not in GENERIC and not re.fullmatch(r"[a-z]{3}[a-z0-9]*", w)) or len(w) > 7 for w in ws if w not in GENERIC) QSTOP = GENERIC | set("""binding atp dna abc duf cassette domain-containing hypothetical family protein subunit class enzyme terminal substrate complex acid dependent dehydro deoxy phospho beta alpha gamma delta atp-binding substrate-binding nad-binding fad-binding dna-binding site""".split()) def groups_line(F, cols, secs, BR): """[Q]: the gene groups whose products agree, found by the server, where a [[ ]] may stand: a run of columns whose most carried genes share a prefix (an operon: leu, cai, fli; waa and rfa are one), a run of columns whose products share a word though their names differ (flagellar: fli, flh, mot), a branch whose families share one. Each with the words most of its families' products carry (in an operon, a third of them and two at least): the words a label may use and still fit the families at those columns.""" sid = {c["k"]: f"S{i}" for i, sec in enumerate(secs, 1) for c in sec["cols"]} def agree(ns, frac): cnt = collections.Counter(w for n in ns for w in sorted(pwords(n, gene=False) - QSTOP) if len(w) >= 4) return [w for w, v in sorted(cnt.items(), key=lambda x: (-x[1], x[0])) if v >= 2 and v >= frac * len(ns)][:4] # ties: a-z lab = lambda n: n["label"] if n.get("named") else None out = [] run = [] # 1. operons: consecutive genes sharing a prefix for c in cols + [None]: n = c["top"] if c else None g = lab(n) if n else None pre = g[:3].lower() if g and re.fullmatch(r"[a-z]{3}[A-Z]\w*", g) and not g.startswith("y") else None pre = {"rfa": "waa"}.get(pre, pre) if run and pre and pre == run[-1][2] and c["k"] == run[-1][0] + 1: run.append((c["k"], n, pre)); continue if len(run) >= 3: ws = agree([x[1] for x in run], 1 / 3) if ws: out.append(dict(a=run[0][0], b=run[-1][0], first=lab(run[0][1]), last=lab(run[-1][1]), words=ws, ids=sorted({sid[x[0]] for x in run}, key=lambda x: int(x[1:])))) run = [(c["k"], n, pre)] if pre else [] taken = set().union(set(), *(set(range(g["a"], g["b"] + 1)) for g in out)) cnt = collections.Counter(w for c in cols for w in sorted(pwords(c["top"], gene=False) - QSTOP) if len(w) >= 5) for w, v in sorted(cnt.items(), key=lambda x: (-x[1], x[0]))[:8]: # 2. a word shared along a run of columns (ties: a-z, # never the set's order, which changes with the process) if v < 3: break hits = [c["k"] for c in cols if w in pwords(c["top"], gene=False)] runs_ = [[hits[0]]] for k in hits[1:]: if k - runs_[-1][-1] <= 2: runs_[-1].append(k) else: runs_.append([k]) for r in runs_: a, b = r[0], r[-1] if len(r) < 3 or len(r) < 0.6 * (b - a + 1) or len(set(range(a, b + 1)) & taken) * 2 > b - a + 1: continue ns = [c["top"] for c in cols if a <= c["k"] <= b] ws = [w] + [x for x in agree(ns, 0.5) if x != w][:3] first, last = next(c["top"] for c in cols if c["k"] == a), next(c["top"] for c in cols if c["k"] == b) out.append(dict(a=a, b=b, first=lab(first), last=lab(last), words=ws, ids=sorted({sid[k] for k in range(a, b + 1) if k in sid}, key=lambda x: int(x[1:])))) taken |= set(range(a, b + 1)) for bid, bb in BR: # 3. a branch whose families agree ws = agree([n for _, n in bb], 0.5) if ws: out.append(dict(a=bb[0][0], b=bb[-1][0], branch=bid, words=ws, ids=[bid])) out = [g for g in out if not any(h is not g and h["a"] <= g["a"] and g["b"] <= h["b"] and not g.get("branch") and set(h["words"]) & set(g["words"]) for h in out)] # a group inside a larger one: one if not out: return out = sorted(sorted(out, key=lambda g: -(g["b"] - g["a"]))[:5], key=lambda g: g["a"]) def say(g): where = f"columns {F.int(g['a'])}-{F.int(g['b'])}" if g.get("branch"): return f"the branch [{g['branch']}] through {where}: {', '.join(g['words'])}" who = f" ({g['first']} to {g['last']})" if g.get("first") and g.get("last") else "" return f"{where}{who} [{', '.join(g['ids'])}]: {', '.join(g['words'])}" F.add("GROUPS whose products agree, the places for a [[ ]] (name that group's columns exactly, cite its line and [Q], make " "one phrase from its words): " + "; ".join(say(g) for g in out) + ".", glob=True, id="Q") def facts(D, wid): """the facts of one window (graph data: pgb/graph_pgb.json for dnaA, pgb_region.build for the others)""" F = Facts(); m = D["meta"]; W = m["window"]; NG = m["n_genomes"]; home = wid == DNAA cols, runs = columns(D); unit = "genomes" if home else "chromosomes"; F.unit = unit F.kinds = {c["k"]: c["kind"] for c in cols} for c in cols: # each gene's share at the columns it is most carried at if c["top"].get("named"): F.gshare[c["top"]["label"]].add(Decimal(fmt_pct(c["s1"]).rstrip("%"))) if home: full = m.get("full_windows") F.add(f"WINDOW: the dnaA window: {F.int(W)} columns from dnaA (column {F.int(0)}) in its direction, in all {F.int(NG)} genomes " f"of the PanGBank E. coli pangenome; {F.int(m['calls'])} calls of the model, each on a genome's real proteins." + (f" {F.int(full)} of the {F.int(NG)} genomes have all {F.int(W)} columns on one contig; in the others the window " f"stops earlier, at a contig end." if full else ""), glob=True, id="W") else: a = m["anchor_label"] if re.fullmatch(r"[a-z]{3}[A-Z]?\w*", a or ""): F.genes.add(a); F.at[a].add(0) pos = m.get("anchor_pos") where = f"; about genes {F.int(pos)} to {F.int(pos + W - 1)} from dnaA" if pos is not None else "" F.add(f"WINDOW: the {a} window: {F.int(W)} columns, column {F.int(0)} is the anchor {a}, then the next {F.int(W - 1)} genes " f"in dnaA's direction{where}. Read in {F.int(NG)} of the {F.int(m['genomes_total'])} complete chromosomes (those " f"carrying {a} once); {F.int(m['calls'])} calls of the model, each on a chromosome's real proteins.", glob=True, id="W") one = unit[:-1] F.add(f"TERMS: share: {unit} carrying the family in the window. named: calls where the model names the gene at that column " f"from the genes before it. bits: the model's entropy there, its hesitation (the more bits, the more hesitant). " f"perplexity (whole window only): how surprised the model is by the real gene. RGP N%: N% of the calls there are on a " f"gene inside a region of genomic plasticity (panRGP). P/S/C: persistent, shell, cloud (PPanGGOLiN). Spine column: " f"one family in {F.pct(0.9)} of {unit} or more; fork: a second family in {F.pct(0.1)} or more; variable: neither. A " f"persistent family is in nearly every {one}: in a fork or variable column its lower share means it lies beyond the " f"window's end in the other {unit} (genes inserted upstream push it out{', or a contig ends' if home else ''}); never " f"call it rare or say they lack it.", glob=True, id="T") pl = D["per_locus"]; n_all = sum(r["n"] for r in pl) or 1 ent = sum(r["entropy"] * r["n"] for r in pl) / n_all part, rg = D.get("by_partition") or {}, D.get("by_rgp") or {} t = lambda r: f"{F.int(r['n'])} calls, named {F.pct(r['top1_dec'])}, {F.bits(r['entropy'])} bits" if r else "no calls" nk = collections.Counter(c["kind"] for c in cols) F.gtop = Decimal(fmt_pct(m["top1_dec"]).rstrip("%")); F.gbits = Decimal(fmt_bits(ent)) F.add(f"WHOLE WINDOW: the model names {F.pct(m['top1_dec'])} of genes, {F.bits(ent)} bits on average (perplexity " f"{F.fam(m['ppl'])} families). By partition of the gene: persistent {t(part.get('persistent'))}; shell " f"{t(part.get('shell'))}; cloud {t(part.get('cloud'))}. Outside an RGP {t(rg.get('outside an RGP'))}; inside an " f"RGP {t(rg.get('inside an RGP'))}. Columns: {F.int(nk['spine'])} spine, {F.int(nk['fork'])} fork, " f"{F.int(nk['variable'])} variable.", glob=True, id="G") RI, RO = rg.get("inside an RGP"), rg.get("outside an RGP") if RI and RO and RI["n"] and RO["n"]: e_i, e_o, a_i, a_o = RI["entropy"], RO["entropy"], RI["top1_dec"], RO["top1_dec"] h = "hesitates more" if e_i > 1.2 * e_o + 0.02 else "hesitates less" if e_i < 0.8 * e_o - 0.02 else "hesitates about as much" n_ = "names fewer genes" if a_i < a_o - 0.03 else "names more genes" if a_i > a_o + 0.03 else "names about as many genes" F.comp = dict(h=h, n=n_, ei=F.bits(e_i), eo=F.bits(e_o), ai=F.pct(a_i), ao=F.pct(a_o)) F.add(f"COMPARISON (computed, over calls): inside an RGP the model {h} than outside ({F.comp['ei']} against {F.comp['eo']} bits) " f"and {n_} ({F.comp['ai']} against {F.comp['ao']}); {F.int(RI['n'])} of the {F.int(RI['n'] + RO['n'])} calls are inside an RGP.", glob=True, id="C") by_k = {c["k"]: c for c in cols} def mean(rs, key): n = sum(r["n"] for r in rs); return sum(r[key] * r["n"] for r in rs) / n if n else None hot = [r for r in pl if r["locus"] in by_k and by_k[r["locus"]]["rgp"] >= 0.5] cold = [r for r in pl if r["locus"] in by_k and by_k[r["locus"]]["rgp"] < 0.5] top = [r for r in sorted(pl, key=lambda r: -r["entropy"])[:3] if r["locus"] in by_k] if pl: F.low_cols.add(min(pl, key=lambda r: r["top1_dec"])["locus"]) # the window's fewest named: its value is in the facts F.hes_cols |= {r["locus"] for r in top} hs = "; ".join(f"column {F.int(r['locus'])} ({F.name(by_k[r['locus']]['top'], k=r['locus'])}, a {by_k[r['locus']]['kind']} column) " f"{F.bits(r['entropy'])} bits, RGP {F.pct(by_k[r['locus']]['rgp'])}" for r in top) def where(rs): """the columns of a short list, as the facts write columns ("columns 9, 36 and 40"; a run as "12-17")""" ks = sorted(r["locus"] for r in rs); runs_ = [] for k in ks: if runs_ and k == runs_[-1][1] + 1: runs_[-1][1] = k else: runs_.append([k, k]) if len(runs_) > 6: return "" it = [F.int(a) if a == b else f"{F.int(a)}-{F.int(b)}" for a, b in runs_] return " (column " + it[0] + ")" if len(it) == 1 and "-" not in it[0] else \ " (columns " + (", ".join(it[:-1]) + " and " + it[-1] if len(it) > 1 else it[0]) + ")" F.add(f"HESITATION: the most hesitant columns of the window: {hs}. " + (f"The {F.int(len(hot))} column{'' if len(hot) == 1 else 's'}{where(hot)} where most calls are in an RGP: named " f"{F.pct(mean(hot, 'top1_dec'))}, {F.bits(mean(hot, 'entropy'))} bits on average; the other {F.int(len(cold))} columns: " f"named {F.pct(mean(cold, 'top1_dec'))}, {F.bits(mean(cold, 'entropy'))} bits." if hot and cold else "No column has most of its calls in an RGP." if not hot else "Every column has most of its calls in an RGP."), glob=True, id="H") sp = [r for r in pl if r["locus"] in by_k and by_k[r["locus"]]["kind"] == "spine"] ot = [r for r in pl if r["locus"] in by_k and by_k[r["locus"]]["kind"] != "spine"] if top: F.hes_spine = by_k[top[0]["locus"]]["kind"] == "spine" if sp and ot: # what the @model paragraph is about: the two kinds of column compared a_s, e_s, a_o, e_o = mean(sp, "top1_dec"), mean(sp, "entropy"), mean(ot, "top1_dec"), mean(ot, "entropy") F.kind = dict(named_spine=a_s, named_other=a_o, bits_spine=e_s, bits_other=e_o) nm = ("more genes on the spine columns than on" if a_s > a_o + 0.02 else "fewer genes on the spine columns than on" if a_s < a_o - 0.02 else "about as many genes on the spine columns as on") hz = ("hesitates less there" if e_s < e_o - 0.02 else "hesitates more there" if e_s > e_o + 0.02 else "hesitates about as much there") pl_ = lambda n, w: f"{F.int(n)} {w} column" + ("" if n == 1 else "s") kt = by_k[top[0]["locus"]]["kind"] if top else None F.add(f"BY KIND OF COLUMN (computed, over the columns with calls): the {pl_(len(sp), 'spine')}: named {F.pct(a_s)}, " f"{F.bits(e_s)} bits on average; the {pl_(len(ot), 'fork and variable')}: named {F.pct(a_o)}, {F.bits(e_o)} bits. " f"The model names {nm} the fork and variable ones, and {hz}." + (f" Its most hesitant column, {F.int(top[0]['locus'])}, is a {kt} column" + (": the model can hesitate most inside a spine." if kt == "spine" else ".") if top else ""), glob=True, id="M") def span(r): return f"column {F.int(r['a'])}" if r["a"] == r["b"] else f"columns {F.int(r['a'])}-{F.int(r['b'])}" def spann(r): return span(r) + (f" ({F.int(r['b'] - r['a'] + 1)} columns)" if r["b"] > r["a"] else "") def rng(vals, f): lo, hi = f(min(vals)), f(max(vals)); return lo if lo == hi else f"{lo.rstrip('%')}-{hi}" def model(cs, listed=True): """the model on a stretch: named (range, mean), where it names fewest, where it hesitates most (or that it hardly does)""" w = [c for c in cs if c["r"]] if not w: return "no calls" n = sum(c["r"]["n"] for c in w); acc = sum(c["r"]["top1_dec"] * c["r"]["n"] for c in w) / n pk = max(w, key=lambda c: c["r"]["entropy"]); lo = min(w, key=lambda c: c["r"]["top1_dec"]) span_ = rng([c['r']['top1_dec'] for c in w], F.pct) s = f"named {span_}" + (f" (mean {F.pct(acc)})" if "-" in span_ else "") who = (lambda c: f" ({F.name(c['top'], k=c['k'])})") if listed else (lambda c: "") if len(w) == 1: return s + f"; {F.bits(pk['r']['entropy'])} bits" hb = pk["r"]["entropy"]; calm = hb < 0.2 if "-" in span_: F.low_cols.add(lo["k"]) if lo is pk and not calm: F.hes_cols.add(pk["k"]) return s + f", fewest named and most hesitant at column {F.int(lo['k'])}{who(lo)}, named {F.pct(lo['r']['top1_dec'])}, {F.bits(hb)} bits" s += f", fewest named at column {F.int(lo['k'])}{who(lo)}, named {F.pct(lo['r']['top1_dec'])}" F.hes_cols.add(pk["k"]) if calm: return s + f"; no column of this stretch hesitates more than {F.bits(hb)} bits (column {F.int(pk['k'])}{who(pk)})" return s + f"; most hesitant at column {F.int(pk['k'])}{who(pk)}, {F.bits(hb)} bits" def parts(cs): c = collections.Counter(PART.get(x["top"]["partition"], "?") for x in cs) return ("all " + next(iter(c))) if len(c) == 1 else ", ".join(f"{F.int(v)} {k}" for k, v in c.most_common()) group_prod = {} # one product for each run of named genes sharing a prefix (an operon, often): its most specific grp = [] for c in cols + [None]: n = c["top"] if c else None pre = n["label"][:3] if n and n.get("named") and not n["label"].startswith("y") and len(n["label"]) >= 4 else None if grp and pre and pre == grp[-1][1]["label"][:3] and c["k"] == grp[-1][0] + 1: grp.append((c["k"], n)); continue if len(grp) >= 2: # the product whose words the group shares most ws = [words(x.get("product", "")) for _, x in grp] sc = [sum(1 + sum(w in o for o in ws) - 1 for w in wi) for wi in ws] j = max(range(len(grp)), key=lambda j: (sc[j], -len(grp[j][1].get("product", "")))) if sc[j]: group_prod[id(grp[j][1])] = grp[j][1]["product"] grp = [(c["k"], n)] if pre else [] def prod(n): """a y-gene's product (its name says little), and one product per group of genes sharing a prefix; an unnamed family is named by its product already""" p = (n.get("product") or "").strip() if not n.get("named") or not p or p.lower() == "hypothetical protein": return "" if n["label"].startswith("y") or id(n) in group_prod: return f' "{F.text(p, 36)}"' return "" def fam_s(n, k, main=None, alt=False): same = " (another family)" if main is not None and n.get("named") and main.get("named") and n["label"] == main["label"] else "" sh = F.pct(n['n'] / NG); nm = F.name(n, k=k, alt=alt) if n.get("named"): F.gshare[n["label"]].add(Decimal(sh.rstrip("%"))) else: F.pshare[nm.strip('"').rstrip("…").lower()].add(Decimal(sh.rstrip("%"))) return f"{nm}{same} {PART.get(n['partition'], '?')} {sh}" def top_s(c): sh = F.pct(c["s1"]) if c["top"].get("named"): F.gshare[c["top"]["label"]].add(Decimal(sh.rstrip("%"))) return f"{F.name(c['top'], k=c['k'])}{prod(c['top'])} {sh}" E = {(e["s"], e["t"]): e["n"] for e in D.get("edges", [])} def branches(cs): """second families of consecutive fork columns that follow one another in most of their genomes: a parallel path""" out, cur = [], [] for c in cs: if cur and c["alts"]: p = cur[-1][1]; nx = max(c["alts"], key=lambda n: E.get((p["id"], n["id"]), 0)) if c["k"] == cur[-1][0] + 1 and E.get((p["id"], nx["id"]), 0) >= 0.5 * min(p["n"], nx["n"]): cur.append((c["k"], nx)); continue if len(cur) >= 3: out.append(cur) cur = [(c["k"], c["alts"][0])] if c["alts"] else [] if len(cur) >= 3: out.append(cur) return out # sections: each spine run, and each run of fork and variable columns between two spines (one header for them all) secs = []; BR = [] for r in runs: if r["kind"] != "spine" and secs and secs[-1]["kind"] == "open" and r["a"] == secs[-1]["b"] + 1: secs[-1]["b"] = r["b"]; secs[-1]["runs"].append(r) else: secs.append(dict(kind="spine" if r["kind"] == "spine" else "open", a=r["a"], b=r["b"], runs=[r])) for sec in secs: sec["cols"] = [c for r in sec["runs"] for c in r["cols"]] kinds = sorted({r["kind"] for r in sec["runs"]}) sec["what"] = "spine" if sec["kind"] == "spine" else {("fork",): "fork", ("variable",): "variable"}.get(tuple(kinds), "fork and variable columns") sec["rgp_mean"] = sum(c["rgp"] for c in sec["cols"]) / len(sec["cols"]) # the stretches most and least in regions of plasticity: the only ones to call so. Among three stretches or more, and # never one that covers most of the window ("the most plastic stretch" of 73 columns out of 80 says nothing) cand = [s for s in secs if len(s["cols"]) * 2 <= len(cols)] if len(secs) >= 3 else [] if len(cand) >= 2: hi = max(cand, key=lambda s: s["rgp_mean"]); lo = min(cand, key=lambda s: s["rgp_mean"]) if hi is not lo and hi["rgp_mean"] > lo["rgp_mean"] + 0.1: allp = min(c["rgp"] for c in lo["cols"]) >= 0.5 # nothing in the window is really outside: "least" says nothing F.plastic = dict(most=(hi["a"], hi["b"]), least=None if allp else (lo["a"], lo["b"])) F.add(f"PLASTICITY (the stretches below, by their mean over columns): the stretch most in regions of plasticity: {span(hi)} " f"({hi['what']}), RGP {rng([c['rgp'] for c in hi['cols']], F.pct)}; " + (f"every stretch of this window lies largely inside regions of plasticity (the lowest, {span(lo)}, RGP " f"{rng([c['rgp'] for c in lo['cols']], F.pct)}): never call a stretch of it the least plastic." if allp else f"the least: {span(lo)} ({lo['what']}), RGP {rng([c['rgp'] for c in lo['cols']], F.pct)}."), glob=True, cols=set(range(hi["a"], hi["b"] + 1)) | set(range(lo["a"], lo["b"] + 1)), id="P") F.add(f"STRETCHES, in window order (each spine; between spines, the fork and variable columns; the family listed is the " f"most carried at its column; share of {unit}; RGP: range over the columns):", glob=True, id="L") for i, sec in enumerate(secs, 1): cs = sec["cols"]; rgp = rng([c["rgp"] for c in cs], F.pct); ks = [c["k"] for c in cs] if sec["kind"] == "spine": genes = ", ".join(F.name(c["top"], k=c["k"]) + prod(c["top"]) for c in cs) F.add(f"{i}. {spann(sec)} spine: share {rng([c['s1'] for c in cs], F.pct)}, {parts(cs)}, RGP {rgp}. Genes: {genes}. {model(cs)}.", cols=ks, id=f"S{i}") continue h = F.add(f"{i}. {spann(sec)} {sec['what']}: share {rng([c['s1'] for c in cs], F.pct)}, {parts(cs)}, RGP {rgp}. " f"{model(cs, listed=False)}.", cols=ks, id=f"S{i}") inb = set() for r in sec["runs"]: for bb in (branches(r["cols"]) if r["kind"] == "fork" else []): ns = [n for _, n in bb]; inb |= {id(n) for n in ns}; BR.append((f"S{i}.b{bb[0][0]}", bb)) F.add(f" a second branch of {F.int(len(ns))} families through columns {F.int(bb[0][0])}-{F.int(bb[-1][0])}, a parallel path " f"that some {unit} carry instead of the most carried families there, each following the previous one in most of its " f"{unit}, share {rng([n['n'] / NG for n in ns], F.pct)}, {parts([dict(top=n) for n in ns])}; the branch's families: " + ", ".join(F.name(n, k=k, alt=True) for k, n in bb), cols=[k for k, _ in bb], head=h, id=f"S{i}.b{bb[0][0]}") mark = (lambda r: f" {r['kind']}") if sec["what"] == "fork and variable columns" else (lambda r: "") for r in sec["runs"]: if r["kind"] == "variable" and len(r["cols"]) > 4: # a long variable run: one line F.add(f" {span(r)}{mark(r)}, share {rng([c['s1'] for c in r['cols']], F.pct)}, {parts(r['cols'])}: " + ", ".join(top_s(c) for c in r["cols"]), cols=[c["k"] for c in r["cols"]], head=h, id=f"S{i}.c{r['cols'][0]['k']}-{r['cols'][-1]['k']}") continue for c in r["cols"]: al = [n for n in c["alts"][:3] if id(n) not in inb] alts = "; also " + ", ".join(fam_s(n, c["k"], c["top"], alt=True) for n in al) if al else "" F.add(f" {F.int(c['k'])}{mark(r)}: {fam_s(c['top'], c['k'])}{prod(c['top'])}{alts}", cols=[c["k"]], head=h, id=f"S{i}.c{c['k']}") groups_line(F, cols, secs, BR) # PRODUCTS: the informative products of named genes in groups sharing a prefix, and of the second families, not shown yet txt = F.txt(); cand = []; tops = [c["top"] for c in cols] for k, n in enumerate(tops): if not n.get("named") or n["label"].startswith("y"): continue if [x for x in tops[max(0, k - 1): k + 2] if x is not n and x.get("named") and x["label"][:3] == n["label"][:3]]: cand.append(n) cand += [a for c in cols for a in c["alts"] if a.get("named")] seen, items = set(), [] for n in cand: lab, p = n["label"], (n.get("product") or "").strip() if lab in seen or not informative(p) or f'{lab} "' in txt: continue seen.add(lab); items.append(f"{lab} {F.text(p, 40)}") if items: F.add("PRODUCTS of named genes (for [[ ]]): " + "; ".join(items[:40]), glob=True, id="X") last = cols[-1] if cols else None if last: F.add(f"END: the window ends at column {F.int(last['k'])}, {F.name(last['top'], k=last['k'])}, in {F.pct(last['s1'])} of {unit}.", cols=[last["k"]], glob=True, id="E") return F # ---------------------------------------------------------------- the prompt STYLE = """**The spine starts at the origin.** Columns 0 to 5, `dnaA`, `dnaN`, `recF`, `gyrB`, `yidB` and `yidA`, sit in essentially every genome, persistent, and hardly ever in a region of plasticity: 0 to 0.5% of calls. The model names `dnaN`, `recF` and `gyrB` in 99.7 to 100% of genomes, `yidB` in 93% and `yidA` in 99%. **The first fork is at `yidX`**, column 6. Then the `dgo` operon, [[galactonate catabolism]], variable columns 7 to 11: about 85% of genomes carry it, and panRGP puts it in a region of plasticity in 90% of them. The model is not lost there: it names each gene in 87 to 99% of genomes, with hardly any hesitation. A variable region can keep a predictable internal order. **Then a long persistent stretch**, `cbrA` to `yicS` through the `ilv` and `uhp` operons, where the model names the next gene in 92% of genomes and seldom hesitates. Around columns 42 to 52 the graph opens: 50% of the calls there fall in a region of plasticity, and the model's hesitation peaks, up to 1.4 bits. Some genomes carry [[type III secretion genes]] there, others [[sugar transport genes]]. **The window ends in the `waa` locus**, [[which builds the lipopolysaccharide core]]: `waaA` in most genomes, then a spread of glycosyltransferases.""" TAGS = dict(start="the first stretch", variable="the middle: fork and variable stretches, branches, gene groups", model="[M], then the most hesitant column of [H]", end="the last stretch and [E]", whole="[G], then [C]") CONTRACT = f"""You write the reading of one window of an E. coli pangenome graph for a scientific web page, from facts the server computes for it: Bacformer, a genomic language model, reads chromosomes gene by gene and at each gene calls the next one; a window is 80 columns of genes. Follow the skill below. FORM. Exactly {len(SECTIONS)} paragraphs in this order, one blank line apart, each opening with its tag and a short bold sentence in **double asterisks**: """ + "; ".join(f"@{t} ({w})" for t, w in TAGS.items()) + f""". Two sentences each (@variable may take three): {WORDS_MIN_ASK} to {WORDS_TARGET} words in all, at most 3 numbers a sentence. Beyond {WORDS_MAX} words whole sentences are cut. CITATIONS. Each facts line begins with its id: [W], [G], [M], [S3], [S3.c12], [S4.b45]. Each sentence ends with the ids of the lines it is written from, in braces, before its full stop: "The model names 93% of them {{S1}}." Every number, gene name, product word and column of a sentence must stand in those lines, or the sentence is dropped. A stretch line [S4] gives its ranges and lets you name the genes, columns and shares of its sub-lines: for columns 15 to 20 of it, their lowest and highest share exactly, or "from A to B%", the first and the last. A branch: its own line and clause. Two or three ids a sentence, {CITES_MAX} at most. RULES. 1. Numbers as the facts write them, in digits (bits may be rounded to 1 decimal); a range as "92 to 99%"; a share, a share in regions of plasticity and a share named are three different numbers. 2. Gene names in backticks; an operon by its prefix (`dgo`); more than 6 in a row as `first` to `last`; an unnamed family by its product, without backticks. 3. Biology only inside [[ ]]: 2 to 6 words from the products of the lines cited, true of most families at the columns named; outside the brackets no product word, function, pathway or cause, not even in bold. 4. Superlatives only as [P], [H] and the stretch lines state them. Write "in a region of plasticity in N% of calls", never "RGP", "share" or the letters P, S, C; spine, fork and variable as the facts label them. Never call a persistent family rare, never say chromosomes lack a gene. 5. [C] and [M] in their direction; never "readily", never "where the graph opens" unless [M] and [H] say so. 6. Plain, exact English; no headings, lists or other markup; nothing about yourself or the facts; no sentence of the reference readings, their windows are not yours.""" _SKILL = [None, None, None] def skill(): """the skill document (reading_skill.md beside this file, READING_SKILL for another path) and its sha, read once: the domain notes, the procedure and the reference reading the model works by. The sha is part of the cache key, so a changed skill makes new readings. Without the file the prompt keeps the reference reading alone.""" if _SKILL[2] is None: try: t = open(SKILL_PATH, encoding="utf-8").read().strip() except OSError: t = "" _SKILL[2] = bool(t) if not t: t = ("# Reading a window of the pangenome graph — the skill document is not on this server\n\nThe reference " "reading, the hand-written one of the dnaA window (its numbers and its biology belong to that window, " f"not to yours):\n\n{STYLE}") _SKILL[0] = t; _SKILL[1] = hashlib.sha1(t.encode()).hexdigest()[:12] return _SKILL[0], _SKILL[1] def system_msg(): t, sha = skill() return f"{CONTRACT}\n\n{t}", sha def user_msg(F, wid): extra = ("\n\nThis is the dnaA window, the one of the first reference reading: write your own reading from these facts; any " "number that is not in the facts is rejected.") if wid == DNAA else "" return (f"Facts, one line each with its id:\n{F.txt()}{extra}\n\nWrite the reading: " + " ".join(f"@{t}" for t in SECTIONS) + ", each sentence ending with the ids of the facts lines it is written from.") # ---------------------------------------------------------------- slips of form, put right without a new call PARTN = {"P": "persistent", "S": "shell", "C": "cloud"} def tidy(text, F): """the model's text with its slips of form put right: no headings or list markers, ranges as "A to B", "RGP" and "share" in words, the letters P/S/C spelled out, a row of more than 6 genes as `first` to `last`""" t = re.sub(r"(?m)^\s*#+\s*", "", text.strip()) t = re.sub(r"(?<=\d)[\u202f\u00a0\u2009](?=\d{3}\b)", ",", t) t = re.sub(r"(?m)^\s*(?:[-•]|\d+\.)\s+", "", t) t = re.sub(r"(?m)^\s*\*\s+(?!\*)", "", t) segs = re.split(r"(`[^`\n]*`)", t) num = r"\d[\d,]*(?:\.\d+)?" for i, s in enumerate(segs): if s.startswith("`") and s.endswith("`") and len(s) > 1: continue s = re.sub(rf"(? 6 else m.group(0) t = re.sub(rf"{G}(?:,\s+{G}){{5,}}(?:,?\s+and\s+{G})?", rows, t) t = re.sub(r"[ \t]{2,}", " ", t) return t.strip() # ---------------------------------------------------------------- the checker NUMW = dict(two=2, three=3, four=4, five=5, six=6, seven=7, eight=8, nine=9, ten=10, eleven=11, twelve=12, thirteen=13, fourteen=14, fifteen=15, sixteen=16, seventeen=17, eighteen=18, nineteen=19, twenty=20, thirty=30, forty=40, fifty=50, sixty=60, seventy=70, eighty=80, ninety=90) NUMW_RE = re.compile(r"\b(" + "|".join(NUMW) + r")\b", re.I) VAGUE_RE = re.compile(r"\b(hundreds?|thousands?|dozens?|half|halves|quarters?|twice|thrice|tenths?|(?:a|one|two)\s+thirds?|" r"(?:nine|eight|seven|six|five|four|three|two|one) (?:in|out of) (?:ten|five|four|three|two))\b", re.I) NUM_RE = re.compile(r"(?= 50: a majority called small (the checker drops it)""" return [dict(kind="claim", text=m.group(0), why="\u201conly\u201d before a majority share") for m in ONLY_RE.finditer(text) if float(m.group(1)) >= 50] HES_RE = re.compile(r"most hesitant|hesitates? (?:the )?most|hesitation peaks|peak hesitation|hesitation is highest|highest hesitation", re.I) LOW_RE = re.compile(r"\bfewest\b|\blowest at\b|least often named|names? (?:the )?fewest|named (?:the )?least", re.I) READY_RE = re.compile(r"\breadily\b|\bwith ease\b|\beasily\b|\bwithout (?:trouble|difficulty|hesitation)\b|\bhas no trouble\b", re.I) OPENS_RE = re.compile(r"where the graph opens|where it opens|where the graph is open|where chromosomes differ|where the order varies|" r"\b(?:in|on|at|across) the (?:fork|variable|plastic|open)\w*|\b(?:in|on|at|across) the fork and variable\b", re.I) PROSE = set("""region regions plasticity model column columns share shares named naming names hesitates hesitation hesitant bits perplexity family families gene genes call calls chromosome chromosomes genome genomes persistent shell cloud spine spines fork forks variable branch branches path paths parallel stretch stretches window windows anchor direction carried carries second first most least other others unnamed average mean readily window's columns' opens""".split()) ALIAS = dict(waa=("rfa",), rfa=("waa",)) OPERON_RE2 = re.compile(r"`([a-z]{3})[A-Z]?[A-Za-z0-9]*`\s+(?:operon|genes|locus|loci|cluster|region)\b") def roundings(v, q0=0): """a fact number and the same number rounded to fewer decimals (at least q0)""" out = {v}; e = -v.as_tuple().exponent for q in range(q0, max(e, 0)): for mode in (ROUND_HALF_UP, ROUND_HALF_EVEN): out.add(v.quantize(Decimal(1).scaleb(-q), rounding=mode)) return out def _allowed(nums): """the numbers a text may write for the facts' numbers: rounded further, but bits to 1 decimal at least (the text's whole-number bits are refused apart), and never to 0% or 100% unless the facts show it""" a = {k: set().union(*(roundings(v, 1 if k == "bits" else 0) for v in vs)) if vs else set() for k, vs in nums.items()} if "pct" in a: a["pct"] = {x for x in a["pct"] if x not in (0, 100) or x in nums["pct"]} return a _LNUM = re.compile(r"(?= a and a not in val and b not in val: cs |= set(range(a, b + 1)) return cs def genes_of(s, F): g = {x.strip() for x in re.findall(r"`([^`\n]+)`", s)} for x in F.genes: if len(x) > 2 and re.search(rf"(?= 6 and p.rstrip("…") in low: cols = cols | ks for g in genes: hit |= {i for i in _gene_lines(g, F, idx) if not idx[i]["glob"]} for a, b in re.findall(r"`([^`]+)`\s+(?:to|through)\s+`([^`]+)`", s): ia = [i for i in _gene_lines(a, F, idx) if not idx[i]["glob"]]; ib = [i for i in _gene_lines(b, F, idx) if not idx[i]["glob"]] if ia and ib: hit |= set(range(min(ia), max(ib) + 1)) for i, L in enumerate(idx): if L["cols"] & cols and not L["glob"]: hit.add(i) heads = {idx[i]["head"] for i in hit if idx[i]["head"] is not None} gcols = set().union(*(F.at.get(g, set()) for g in genes)) if genes else set() return hit | heads, cols | gcols def check(text, F): """the whole-text check (r7), kept for comparison with the per-sentence one (pg_reading.py check prints both, and it reads a text that carries no citations) -> dict(ok, numbers, names, words, bad=[...], hints=[...]): every number, gene name, number word and identifier with a digit of text, each number against the facts lines of its sentence's genes and columns, and the claims the facts make. Nothing ties a number to the line the sentence is written from: that is what check_cited does.""" ftxt = F.txt(); flow = ftxt.lower(); idx = _line_index(F) allowed = _allowed(F.nums) anyv = set().union(*allowed.values()) if allowed else set() bad = []; nnum = nname = 0 body = text for fr in sorted(F.frags, key=len, reverse=True): # "family 4" of a product: a name, not a number body = re.sub(re.escape(fr), lambda mm: re.sub(r"\d", "#", mm.group(0)), body, flags=re.I) interp_spans = [mm.span() for mm in re.finditer(r"\[\[.*?\]\]", body, flags=re.S)] # identifiers with a digit and letters (IS3, 16S, DUF1234, 2-dehydro-3-deoxy...) must be in the facts; inside [[ ]] a # chemical name with a locant (2-oxoglutarate) is allowed def ident(mm): tok = mm.group(0).strip("'-/.") if not re.search(r"[A-Za-z]", tok): return mm.group(0) u = re.fullmatch(r"(\d[\d,]*(?:\.\d+)?)-(bits?|fold|columns?|genes?|famil(?:y|ies)|percent)", tok) if u: return u.group(1) + " " + u.group(2) # 0.4-bit: a number and its unit inside = any(a <= mm.start() < b for a, b in interp_spans) chem = re.fullmatch(r"\d+(?:,\d+)*-[A-Za-z]{4,}[\w,-]*", tok) and not re.match(r"\d+(?:,\d+)*-(?:genes?|columns?|famil|fold|bits?|stretch)", tok) if tok.lower() not in flow and not (inside and chem): bad.append(dict(kind="identifier", text=tok, why="not in the facts")) return re.sub(r"\d", "#", mm.group(0)) body = IDENT_RE.sub(ident, body) for mm in NUM_RE.finditer(body): raw = mm.group(0); nnum += 1 try: v = Decimal(raw.replace(",", "")) except InvalidOperation: continue u = UNIT_RE.match(body, mm.end()); unit = (u.group(1).lower() if u else "") if not unit: # a range's lower end: the unit of its upper end lo = LOWER_RE.match(body, mm.end()) if lo: unit = lo.group(2).lower() kind = "pct" if unit in ("%", "percent", "per cent") else "bits" if "bit" in unit else "fam" if unit.startswith("famil") and "." in raw else None ok = v in (allowed.get(kind, set()) if kind else anyv) if ok and kind == "bits" and "." not in raw and v not in F.nums["bits"]: bad.append(dict(kind="format", text=f"{raw} bits", why="bits rounded to a whole number: write them as the facts do")); continue if not ok: ctx = body[max(0, mm.start() - 30): mm.end() + 12].replace("\n", " ") bad.append(dict(kind="number", text=raw + ("%" if kind == "pct" else f" {unit}" if unit else ""), why=f"not a {dict(pct='percent', bits='bits value', fam='families value').get(kind, 'number')} of the facts", context=ctx.strip())) plain = re.sub(r"\[\[.*?\]\]", " ", text, flags=re.S) for mm in re.finditer(r"(?= 0 else 10 ** 6 in_first = fi < low.find("outside") if "outside" in low else True if "against" not in low and "outside" in low and "inside" in low and in_first: seg_in = s[low.find("inside"): low.find("outside")]; seg_out = s[low.find("outside"):] for mm in re.finditer(r"(?= 7} out = [] for m in re.finditer(r"\[\[(.+?)\]\]", text): for w in re.findall(r"[a-z]{4,}", m.group(1).lower()): if w in STOP or w in GENERIC_OK or w in V or (len(w) >= 7 and w[:7] in P): continue out.append(dict(text=w, where=m.group(0))) return out # ---------------------------------------------------------------- the sections and the citations (one per sentence) ROLE_KIND = dict(share="pct", rgp="pct", named="pct", definition="pct", bits="bits", ppl="fam", count="int") ROLE_SAY = dict(share="a share of {unit}", rgp="a share in a region of plasticity", named="a share of genes named", bits="a hesitation in bits", ppl="a perplexity", count="a count", definition="a number of the TERMS line") _FNUM = re.compile(r"(?= 7 else w _SUF = sorted(("ases", "ase", "ysis", "oses", "osis", "ations", "ation", "isation", "ization", "ising", "izing", "ing", "ers", "er", "ors", "or", "ions", "ion", "ive", "ic", "al", "ed", "es", "s"), key=len, reverse=True) def _norm(w): """a product word cut back to its stem: hydrolysis and glycosylhydrolase both give hydrol, secretion and secreted both secret, so a word of an interpretation is recognised in the product it comes from""" w = w.lower().strip("'-") for suf in _SUF: if w.endswith(suf) and len(w) - len(suf) >= 5: return w[:len(w) - len(suf)] return w def _same(a, b): """two words of the same thing: equal, or one's stem inside the other (never two words that merely start alike, so transport does not match transposase)""" na, nb = _norm(a), _norm(b) return na == nb or (len(na) >= 6 and na in nb) or (len(nb) >= 6 and nb in na) def _covers(w, pool): return any(_same(w, x) for x in pool) def _role_left(s, rules): """the role the last keyword before a number gives it (None: no keyword)""" best, role = -1, None for rx, r in rules: for m in rx.finditer(s): if m.start() > best: best, role = m.start(), r return role def _cite_index(F): """per facts line, by its id: the words it shows, its numbers by role (share, rgp, named, bits, ppl, count), its columns, and whether it speaks of the whole window. The role of a number is the last keyword before it (RGP, share, named, mean), else its unit, else a share: what the facts lines are built to say.""" if getattr(F, "_cidx", None): return F._cidx out = {} for lid, ln, inf in zip(F.ids, F.lines, F.info): roles = collections.defaultdict(set); prev = 0; terms = ln.startswith("TERMS"); got = [] for m in _FNUM.finditer(ln): unit = (m.group(3) or "").lower(); left = ln[prev:m.start()]; prev = m.end() soft = False if unit.startswith("bit"): role = "bits" elif unit.startswith("famil"): role = "ppl" if re.search(r"perplexity", left, re.I) else "count" elif unit == "%": role = "definition" if terms else _role_left(left, _FROLE) soft = role is None; role = role or "share" else: role = "count"; soft = True got.append([role, [m.group(1), m.group(2)], _JOIN.fullmatch(left) is not None, "." in (m.group(1) or ""), soft]) for j, g in enumerate(got): # "78% against 94%", "0.83 against 0.21 bits": one pair, one role if g[4] and g[2] and j and got[j - 1][0] != "count": g[0] = got[j - 1][0] elif g[4] and g[3] and j + 1 < len(got) and got[j + 1][2] and got[j + 1][0] != "count": g[0] = got[j + 1][0] spans = collections.defaultdict(list); colline = bool(re.fullmatch(r"S\d+\.c\d+", lid)) for role, gs, _j, _d, _s in got: for g in gs: if not g: continue try: roles[role].add(Decimal(g.replace(",", ""))) except InvalidOperation: pass if role in ("share", "rgp", "named") and not (colline and spans["share"]): try: spans[role].append(tuple(sorted(Decimal(g.replace(",", "")) for g in (gs[0], gs[1] or gs[0])))) except InvalidOperation: pass colvals = {} # the share at each column the line lists, in order mr = re.fullmatch(r"S\d+\.c(\d+)-(\d+)", lid) if colline and spans.get("share"): colvals[int(lid.split(".c")[1])] = spans["share"][0][0] elif mr and ":" in ln: vs = [Decimal(x) for x in re.findall(r"(? [S4.c34], [S4.c53-79], [S4.b45] p = lid.split(".")[0] if p != lid and p in out: out[p]["subs"].append(lid) F._cidx = out; return out def parse(text): """the model's answer as its sections: [dict(tag, sents=[dict(text, punct, cites, bold)])], each sentence what ends at a citation in braces -> (sections, dict(tail, untagged, dup, order)): what stands outside a section or after the last citation of one is a fragment the reading cannot use""" secs = []; notes = dict(tail=[], untagged="", dup=[], tags=[]) parts = TAG_RE.split("\n" + text.strip()) if parts[0].strip(): notes["untagged"] = " ".join(parts[0].split())[:200] for tag, body in zip(parts[1::2], parts[2::2]): body = " ".join(body.split()) notes["tags"].append(tag) if any(s["tag"] == tag for s in secs): notes["dup"].append(tag); continue sents = []; pos = 0 for m in CITE_RE.finditer(body): t = body[pos: m.start()].strip(); pos = m.end() ids = CID_RE.findall(m.group(1)) if not t: continue p = m.group(2) or "" if t[-1] in ".;:!?": p = "" elif not p: p = "." sents.append(dict(text=t, punct=p, cites=ids, bold=bool(re.search(r"\*\*.+?\*\*", t)))) rest = body[pos:].strip() if len(rest) > 3: notes["tail"].append(rest[:160]) if sents: secs.append(dict(tag=tag, sents=sents)) notes["order"] = [s["tag"] for s in secs] return secs, notes def _sent_bad(s, ids, F, X, extra=()): """what a sentence says that the facts lines it cites do not carry -> (bad, numbers, names): its numbers by role, its identifiers, its gene names, its columns, the words inside [[ ]], and the claims the facts make elsewhere. A stretch line carries its own sub-lines for gene names, products and columns, never for a number: a number belongs to the line that states it, and a stretch's own line states its ranges. extra: the other sentences' ids, for the sentence that opens a paragraph — an opener may summarise what its paragraph states.""" bad = []; lines = [] for i in ids: got = _resolve_id(i, X) if not got: bad.append(dict(kind="citation", text=i, why="no facts line has this id", context=s[:120])) lines += [L for L in got if L not in lines] if len(ids) > CITES_MAX: # a sentence citing everything is a sentence citing nothing bad.append(dict(kind="citation", text=", ".join(ids), why=f"a sentence cites at most {CITES_MAX} lines", context=s[:120])) if not lines: return bad + [dict(kind="citation", text=s[:60], why="the sentence cites no facts line", context=s[:120])], 0, 0 inh = [L for i in dict.fromkeys(extra) for L in _resolve_id(i, X) if L not in lines] main = lines + inh subs = [X[j] for L in main for j in L["subs"] if X[j] not in main] roles = collections.defaultdict(set); wlow = set(); cols = set(); low = [] for L in main: for k, v in L["roles"].items(): roles[k] |= v hd = X.get(L["id"].split(".")[0]) # a stretch of one column: its line and its column's if hd is not None and hd is not L and hd["cols"] == L["cols"]: # line speak of the same column for k, v in hd["roles"].items(): roles[k] |= v vis = set() for L in main + subs: wlow |= L["words"] | L["show"]; low.append(L["low"]) if L["id"] not in ("Q", "X"): cols |= L["cols"] vis |= set().union(set(), *(f["words"] for f in L["fams"])) byhead = collections.defaultdict(set) # two column lines of one stretch cited: the columns for L in main: # between them are that stretch's too if "." in L["id"] and L["cols"]: byhead[L["id"].split(".")[0]] |= L["cols"] for h, cs in byhead.items(): if h in X and len(cs) >= 1 and sum(1 for L in main if L["id"].startswith(h + ".")) >= 2: cols |= {c for c in X[h]["cols"] if min(cs) <= c <= max(cs)} heads = [X[h] for h in {L["id"].split(".")[0] for L in main} if h in X] vis |= wlow low = " ".join(low) cited = ", ".join(L["id"] for L in lines) pool = collections.defaultdict(set) for r, vs in roles.items(): pool[ROLE_KIND[r]] |= vs br = [L for L in lines if re.search(r"\.b\d+$", L["id"])] # a branch spoken of: its own families, its own shares, bm = re.search(r"\bbranch(?:es)?\b|\bparallel path\b", s, re.I) # in the clause that speaks of it (from the clause bmode = bool(br) and bool(bm) and len(lines) > len(br) # before "branch" to the end of the sentence) bstart = max(s.rfind(",", 0, bm.start()), s.rfind(";", 0, bm.start()), s.rfind(":", 0, bm.start())) + 1 if bmode else 0 if bmode and "[[" in s[bstart: bm.start()] or bmode and s[:bstart].count("[[") != s[:bstart].count("]]"): bstart = 0 inb = lambda i: bmode and i >= bstart if bmode: bwords = set().union(*(L["words"] | L["show"] for L in br)) broles = collections.defaultdict(set) for L in br: for k, v in L["roles"].items(): broles[k] |= v nnum = nname = 0 body = s for fr in sorted(F.frags, key=len, reverse=True): # "family 4" of a product: a name, not a number body = re.sub(re.escape(fr), lambda mm: re.sub(r"\d", "#", mm.group(0)), body, flags=re.I) spans = [mm.span() for mm in re.finditer(r"\[\[.*?\]\]", body, flags=re.S)] inside = lambda i: any(a <= i < b for a, b in spans) def ident(mm): tok = mm.group(0).strip("'-/.") if not re.search(r"[A-Za-z]", tok): return mm.group(0) u = re.fullmatch(r"(\d[\d,]*(?:\.\d+)?)-(bits?|fold|columns?|genes?|famil(?:y|ies)|percent)", tok) if u: return u.group(1) + " " + u.group(2) # 0.4-bit: a number and its unit chem = re.fullmatch(r"\d+(?:,\d+)*-[A-Za-z]{4,}[\w,-]*", tok) and not re.match(r"\d+(?:,\d+)*-(?:genes?|columns?|famil|fold|bits?|stretch)", tok) if tok.lower() not in low and not (inside(mm.start()) and chem): bad.append(dict(kind="identifier", text=tok, why=f"not in the lines cited ({cited})", context=s[:120])) return re.sub(r"\d", "#", mm.group(0)) body = IDENT_RE.sub(ident, body) rsp = [(m.start(1), m.end(1), m.start(2), m.end(2)) for m in # the two ends of a range re.finditer(r"(?= 6 and p_[:14] in pk] if not g_ and pf: # "an IS1-like element transposase in 29%": that family's if v not in set().union(*(_allowed({"pct": {y}})["pct"] for x in pf for y in x)): bad.append(dict(kind="number", text=f"{raw}%", why="not the share of the family it follows in the facts", context=s[:120])); continue elif not g_: ccs = _clause_cols(s, mm.start(), F) # "column 10 ... carried by 64%": that column's share if len(ccs) == 1 and min(ccs) in cvals and not re.search(r"\bgenomes\b|\bchromosomes\b\s*$", pre_[-3:]) \ and v not in _allowed({"pct": {cvals[min(ccs)]}})["pct"] and re.search(r"carri|share|\bin\s*$|\bat\s*$", pre_[-24:]): bad.append(dict(kind="number", text=f"{raw}%", why=f"column {min(ccs)} is carried by {cvals[min(ccs)]}% in the lines cited " f"({cited})", context=s[:120])); continue if ok and kind == "pct" and role == "named" and F.gtop is not None and any(L["id"] == "G" for L in lines) \ and re.search(r"names?\s+$", body[max(0, mm.start() - 12): mm.start()]) and re.match(r"%\s+of\s+(?:the\s+)?genes\b", body[mm.end():]) \ and not re.search(r"persistent|shell|cloud|inside|outside|spine|fork|variable", re.split(r"[.;]", s[:mm.start()])[-1], re.I): if v not in _allowed({"pct": {F.gtop}})["pct"]: # "across the window it names 89% of genes": [G]'s own bad.append(dict(kind="number", text=f"{raw}%", why=f"across the window the model names {F.gtop}% of genes [G]", context=s[:120])); continue if ok and kind == "bits" and F.gbits is not None and any(L["id"] == "G" for L in lines) and not any(L["id"] in ("H", "M") for L in lines) \ and re.match(r"\s*bits?\s+on\s+average", body[mm.end():]) and v not in _allowed({"bits": {F.gbits}})["bits"]: bad.append(dict(kind="number", text=f"{raw} bits", why=f"across the window it hesitates {F.gbits} bits on average [G]", context=s[:120])); continue if ok and kind == "bits" and "." not in raw and v not in roles.get("bits", set()): bad.append(dict(kind="format", text=f"{raw} bits", why="bits rounded to a whole number: write them as the facts do", context=s[:120])); continue if not ok: say = ROLE_SAY.get(role, "a number").format(unit=F.unit) if role else "a number" bad.append(dict(kind="number", text=raw + ("%" if kind == "pct" else f" {unit}" if unit else ""), why=f"not {say} of the lines cited ({cited})", context=s[:120])) for mm in re.finditer(r"(? b_) if down else (a_ < b_) for a_, b_ in froms): bad.append(dict(kind="claim", text=mt.group(0), why="a trend along the columns needs its values: \"from A to B%\" at the first " "and last column named, from the lines cited", context=s[:120])) allv = set().union(*pool.values()) if pool else set() for mm in NUMW_RE.finditer(s): nnum += 1 if Decimal(NUMW[mm.group(1).lower()]) not in allv: bad.append(dict(kind="number word", text=mm.group(0), why="write numbers in digits, from the lines cited")) for mm in VAGUE_RE.finditer(s): bad.append(dict(kind="number word", text=mm.group(0), why="write numbers in digits, from the lines cited")) for mm in META_RE.finditer(s): bad.append(dict(kind="wording", text=mm.group(0), why="the reading speaks of the window, not of the facts")) whyg = (f"a branch is spoken of with its own families only ({', '.join(L['id'] for L in br)})" if bmode else f"not in the lines cited ({cited})") def known(g, at=0): gl = g.lower(); b = inb(at) pool_ = bwords if b else wlow if gl in pool_: return True if len(gl) == 3 and any(w.startswith(gl) for w in pool_ if w in {x.lower() for x in F.genes}): return True if b: return False return gl in low or any(k.endswith("…") and gl.startswith(k[:-1]) for k in F.atp) whyn = f"not in the lines cited ({cited})" for mm in re.finditer(r"`([^`\n]+)`", s): g = mm.group(1).strip(); nname += 1 if not known(g, mm.start()): bad.append(dict(kind="gene", text=g, why=whyg if inb(mm.start()) else whyn, context=s[:120])) nog = re.sub(r"`[^`\n]*`", " ", s) for mm in GENE_RE.finditer(nog): g = mm.group(1); nname += 1 if not known(g, mm.start()) and g not in ALLOWED_WORDS: bad.append(dict(kind="gene", text=g, why=whyg if inb(mm.start()) else whyn, context=s[:120])) for mm in PROT_RE.finditer(nog): g = mm.group(1) if not known(g, mm.start()) and not known(g[0].lower() + g[1:], mm.start()) and g not in ALLOWED_WORDS: nname += 1; bad.append(dict(kind="gene", text=g, why=whyg if inb(mm.start()) else whyn, context=s[:120])) for mm in OPERON_RE.finditer(re.sub(r"\[\[.*?\]\]", " ", nog)): g = mm.group(1) if not known(g, mm.start()) and g.lower() not in WORDS3: bad.append(dict(kind="gene", text=g, why=f"not an operon of the lines cited ({cited})", context=s[:120])) for mm in OPERON_RE2.finditer(s): # "the `waa` locus" over a stretch: most of its families pre = mm.group(1).lower() # must carry that name, or the label covers what it is not oc = _clause_cols(s[:mm.start()] + s[mm.end():], mm.start(), None) # the columns named nearest, not the operon's own genes fs = [f for L in main + subs if not L["glob"] for f in L["fams"] if not f["alt"] and (not oc or f["col"] is None or f["col"] in oc)] pres = {pre} | set(ALIAS.get(pre, ())) # rfa is the old name of waa nin = sum(f["label"].lower().startswith(tuple(pres)) for f in fs) if len(fs) >= 4 and nin * 2 < len(fs): bad.append(dict(kind="claim", text=mm.group(0), why=f"only {nin} of the {len(fs)} families there carry that name: name " f"the columns it is true of", context=s[:120])) miss = sorted(cols_of(s) - cols) if miss: bad.append(dict(kind="column", text="column " + ", ".join(map(str, miss[:6])), why=f"the lines cited ({cited}) do not speak of that column", context=s[:120])) scols = cols_of(s) for mm in re.finditer(r"\[\[(.+?)\]\]", s): nw = len(re.findall(r"[A-Za-z0-9][\w'-]*", mm.group(1))) if nw < 2 or nw > 6 or re.search(r"[,;]", mm.group(1)): bad.append(dict(kind="interpretation", text=mm.group(1)[:40], why="an interpretation is one phrase of 2 to 6 words, not a word " "or a list", context=mm.group(0)[:80])) cw = [w for w in re.findall(r"[A-Za-z][\w'-]{3,}|\b[A-Z][A-Za-z0-9]{1,2}\b", mm.group(1)) # GMP, DNA, Fe: words too if w.lower() not in STOP and w.lower() not in ("the", "and", "for", "its")] for w in cw: # every word of the label: in the lines cited if _covers(w.lower(), vis): continue bad.append(dict(kind="interpretation", text=w, why=f"not a word of the products or gene names of the lines cited ({cited})", context=mm.group(0)[:80])) if not cw: # "[[sugar transport]]": nothing the products must say bad.append(dict(kind="interpretation", text=mm.group(1)[:40], why="it names nothing the products of the lines cited say: build " "it from their words", context=mm.group(0)[:80])); continue fams, seen_f = [], set(); scols = _clause_cols(s, mm.start(), F) # the families the label is about: those of the b_ = (inb(mm.start()) or bstart == 0) if bmode else bool(br) and len(br) == len(lines) # stretch, column or branch nb_ = [x for x in main + subs if not x["glob"] and not (bmode and x in br)] # lines cited, narrowed to the columns named for L in (br if b_ else nb_): for f in L["fams"]: k = (f["label"], f["col"]) if k in seen_f or (f["alt"] and not b_) or (scols and f["col"] is not None and f["col"] not in scols and not b_): continue seen_f.add(k); fams.append(f) if not fams: for g in genes_of(s, F): for L in X.values(): for f in L["fams"]: if f["label"] == g and (f["label"], f["col"]) not in seen_f: seen_f.add((f["label"], f["col"])); fams.append(f) if not fams: bad.append(dict(kind="interpretation", text=mm.group(1)[:40], why=f"the lines cited ({cited}) name no family this could be about", context=mm.group(0)[:80])); continue hit = [f for f in fams if any(_covers(w.lower(), f["words"]) for w in cw)] pre = {f["label"][:3].lower() for f in fams if re.fullmatch(r"[a-z]{3}[A-Z]\w*", f["label"] or "")} operon = len(pre) == 1 and all(re.fullmatch(r"[a-z]{3}[A-Z]\w*", f["label"] or "") for f in fams) fits = len(hit) >= 2 and len(hit) * 3 >= len(fams) if operon else len(hit) * 2 >= len(fams) if len(fams) >= 3 and not fits: # a label must fit most of the families it covers; in one bad.append(dict(kind="interpretation", text=mm.group(1)[:40], why=f"this says what {len(hit)} of the {len(fams)} families of the lines cited ({cited}) do: name the columns " f"it is true of, or leave the stretch unnamed", context=mm.group(0)[:80])) else: parts_ = [[w for w in re.findall(r"[A-Za-z][\w'-]{3,}", x) if w.lower() not in STOP and w.lower() not in GENERIC and w.lower()[:-1] not in GENERIC] for x in re.split(r"\s+and\s+|,\s*|\s*/\s*", mm.group(1))] for cp in [x for x in parts_ if len(x) >= 2] if len(fams) >= 3 else []: if not any(sum(_covers(w.lower(), f["words"]) for w in cp) >= 2 for f in hit): bad.append(dict(kind="interpretation", text=" ".join(cp)[:40], why="its words come one from each family: no family of the lines cited does all of it", context=mm.group(0)[:80])); break for rx, what, key, unit_ in ((HES_RE, "most hesitant", "hesv", r"(\d+(?:\.\d+)?)\s*bits?\b"), (LOW_RE, "fewest named", "lowv", r"(\d+(?:\.\d+)?)\s*%")): for mm in rx.finditer(s): if what == "most hesitant" and re.search(r"\bno column\b|hardly|barely|seldom", s, re.I): continue st_ = [t for L in main for t in L[key]] # what the lines cited state, column and value if not st_: bad.append(dict(kind="claim", text=mm.group(0), why=f"the lines cited ({cited}) do not state a column as the {what}", context=s[:120])); continue cc = _claim_cols(s, mm, F) or cols_of(s) if cc and not (cc & {k2 for k2, _ in st_}): bad.append(dict(kind="claim", text=mm.group(0), why=f"the lines cited ({cited}) state column {', '.join(str(k2) for k2, _ in st_[:3])} " f"as the {what}", context=s[:120])); continue seg = re.split(r"[;]|\.\s", s[mm.end():])[0] vm = re.search(unit_, seg) if not vm: bef = list(re.finditer(unit_, s[:mm.start()])); vm = bef[-1] if bef else None if vm: # its value: the one stated with that column vv = Decimal(vm.group(1)); okv = [x for k2, x in st_ if (not cc or k2 in cc)] if okv and not any(vv in _allowed({"bits" if key == "hesv" else "pct": {x}})["bits" if key == "hesv" else "pct"] for x in okv): bad.append(dict(kind="number", text=vm.group(0), why=f"the {what} there is {okv[0]}{' bits' if key == 'hesv' else '%'} in the lines " f"cited ({cited})", context=s[:120])) for mm in PLASTIC_RE.finditer(s): ab = F.plastic.get("most" if mm.group(1).lower() == "most" else "least") if not ab or not any(L["id"] == "P" for L in lines): bad.append(dict(kind="claim", text=mm.group(0), why="only the PLASTICITY line [P] names a stretch so", context=s[:120])); continue cc = cols_of(s) if cc and not (cc & set(range(ab[0], ab[1] + 1))): bad.append(dict(kind="claim", text=mm.group(0), why=f"the facts say so of columns {ab[0]} to {ab[1]} only", context=s[:120])) plain = re.sub(r"\[\[.*?\]\]", " ", s) K, ready, opens = F.kind, READY_RE.search(plain), OPENS_RE.search(plain) hesit = re.search(r"\bhesitat", plain, re.I) mcited = any(L["id"] == "M" for L in main) if mcited and K: bad += _m_direction(plain, K, s) # the two kinds of column compared, as [M] states it nbh = [X[f"S{int(h['id'][1:]) + d}"] for h in heads if re.fullmatch(r"S\d+", h["id"]) for d in (-1, 1) if f"S{int(h['id'][1:]) + d}" in X] # "breaks the spine": the stretches on either side for kw in ("spine", "fork", "variable", "branch"): # a kind of column the sentence speaks of: in its lines if re.search(rf"\b{kw}(?:e?s)?\b", plain, re.I) and not any(kw in L["low"] for L in main + subs + heads + (nbh if kw != "branch" else [])): bad.append(dict(kind="claim", text=kw, why=f"no line cited ({cited}) has a {kw} here", context=s[:120])) for mm in re.finditer(r"(?= i else i - mm.end() if best is None or d < best[0]: best = (d, c) return best[1] if best else cols_of(s[a:b]) or cols_of(s) TREND_DOWN = re.compile(r"\bfewer and fewer\b|\b(?:falling|decreasing|declining|dwindling|dropping|thinning)\b", re.I) TREND_UP = re.compile(r"\bmore and more\b|\b(?:rising|increasing|growing)\b", re.I) def _ends(cs, lines): """the shares the lines show at the first and last of the columns cs (None when one of them is not shown)""" if not cs: return None cv = {} for L in lines: for k, v in L.get("colvals", {}).items(): cv.setdefault(k, v) a, b = min(cs), max(cs) return (cv[a], cv[b]) if a in cv and b in cv and a != b else None def _range_ok(a, b, role, lines, scols, X): """a range of a sentence ("84 to 88%"): one a line cited shows; or the lowest and highest the lines cited show, in that role (several column lines); or, for a share, the values at the first and last column the sentence names""" al = lambda x: _allowed({"pct": {x}})["pct"] lo_, hi_ = min(a, b), max(a, b) sp = [t for L in lines for t in L.get("spans", {}).get(role, [])] if not sp: return True # nothing to hold it against: the value check decides if any(lo_ in al(x) and hi_ in al(y) for x, y in sp): return True covered = set().union(set(), *(L["cols"] for L in lines if L.get("spans", {}).get(role))) if lo_ in al(min(x for x, _ in sp)) and hi_ in al(max(y for _, y in sp)) and (not scols or scols <= covered): return True if role == "share" and scols: ends = [] for k in (min(scols), max(scols)): L = next((L for L in lines if re.fullmatch(r"S\d+\.c\d+", L["id"]) and L["cols"] == {k}), None) if L is None or not L["spans"].get("share"): return False ends.append(L["spans"]["share"][0][0]) if {x for e in ends for x in al(e)} >= {a, b} and a in al(ends[0]) and b in al(ends[1]): return True return False def _resolve_id(i, X): """the lines an id stands for: itself; a range the model made up ("S2.c77-79") or one column of a range line ("S2.c20" for [S2.c20-24]): the column lines of that stretch within it""" if i in X: return [X[i]] m = re.fullmatch(r"(S\d+)\.c(\d+)(?:-(\d+))?", i) if not m or m.group(1) not in X: return [] a = int(m.group(2)); b = int(m.group(3) or a) out = [X[j] for j in X[m.group(1)]["subs"] if re.match(r"S\d+\.c", j) and X[j]["cols"] and (min(X[j]["cols"]) <= b and max(X[j]["cols"]) >= a)] return out if out and set().union(*(L["cols"] for L in out)) >= set(range(a, b + 1)) & X[m.group(1)]["cols"] else [] _KIND_W = r"(spines?|spine columns?|fork|variable|others?|other ones)" def _m_direction(p, K, s): """"names more genes on the spine columns", "hesitates less there": in the direction [M] gives, for the kind of column the clause is about (said, or the first one the sentence names)""" bad = []; low = p.lower() kinds = [(low.find(w), w) for w in ("spine", "fork", "variable") if w in low] first = min(kinds)[1] if kinds else None side = lambda w: "spine" if w and w.startswith("spine") else "other" if w else None for m in re.finditer(r"\bnames?\s+(?:about\s+|nearly\s+|roughly\s+)?(more|fewer|as many)\s+(?:genes\s+)?(?:(?:on|in|at|over)\s+the\s+" + _KIND_W + r"|there)?", low): sd = side(m.group(2)) or side(first) if not sd: continue d = K["named_spine"] - K["named_other"]; d = d if sd == "spine" else -d want = "more" if d > 0.02 else "fewer" if d < -0.02 else "as many" if m.group(1) != want: bad.append(dict(kind="claim", text=m.group(0), why=f"[M]: the model names {'about as many' if want == 'as many' else want} genes " f"on the {'spine' if sd == 'spine' else 'fork and variable'} columns", context=s[:120])) for m in re.finditer(r"\bhesitat\w*\s+(?:about\s+|nearly\s+)?(more|less|as much)\b(?:\s+(?:(?:on|in|at|over)\s+the\s+" + _KIND_W + r"))?", low): sd = side(m.group(2)) or ("other" if OPENS_RE.search(low[m.end():]) else side(first)) if not sd: continue d = K["bits_spine"] - K["bits_other"]; d = d if sd == "spine" else -d want = "more" if d > 0.02 else "less" if d < -0.02 else "as much" if m.group(1) != want: bad.append(dict(kind="claim", text=m.group(0), why=f"[M]: the model hesitates {'about as much' if want == 'as much' else want} " f"on the {'spine' if sd == 'spine' else 'fork and variable'} columns", context=s[:120])) return bad def check_cited(text, F): """the model's answer against the facts, sentence by sentence, each against the lines it cites (stricter than the whole-text check: nothing may come from another line) -> dict(ok, sections, notes, bad, numbers, names, words, sentences, cited, structure)""" X = _cite_index(F) secs, notes = parse(text) nnum = nname = 0; bad = []; ns = 0 for sec in secs: rest = [i for st in sec["sents"][1:] for i in st["cites"]] # an opener may summarise its own paragraph for i, st in enumerate(sec["sents"]): st["inherit"] = [x for x in dict.fromkeys(rest) if x not in st["cites"]] if i == 0 else [] b, n1, n2 = _sent_bad(st["text"], st["cites"], F, X, st["inherit"]) st["bad"] = b; st["ok"] = not b; st["opens"] = i == 0 nnum += n1; nname += n2; ns += 1; bad += b for sec in secs: # "the spine after it", "then": further along the window prev = None for i, st in enumerate(sec["sents"]): cs = cols_of(st["text"]) or set().union(set(), *[L["cols"] for c in st["cites"] for L in _resolve_id(c, X) if not L["glob"]]) if prev and cs and i and re.search(r"\b(?:after (?:it|them|this|that)|then|next|further on|beyond (?:it|them))\b", re.sub(r"\[\[.*?\]\]", " ", st["text"]), re.I) and min(cs) < min(prev): b = dict(kind="claim", text="after it", why=f"columns {min(cs)} to {max(cs)} come before the columns of the sentence " f"before (from {min(prev)}): the window is read in its order", context=st["text"][:120]) st["bad"].append(b); st["ok"] = False; bad.append(b) prev = cs or prev order = [s["tag"] for s in secs] want = [t for t in SECTIONS if t in order] struct = [] if order != want: struct.append(f"the paragraphs are {' '.join('@' + t for t in order) or 'not tagged'}: write " + " ".join("@" + t for t in SECTIONS) + " in that order") for t in SECTIONS: if t not in order: struct.append(f"the @{t} paragraph is missing") if notes["dup"]: struct.append("these paragraphs are written twice: " + ", ".join("@" + t for t in notes["dup"])) if notes["untagged"]: struct.append(f'this stands before the first paragraph, with no tag: "{notes["untagged"][:80]}"') for t in notes["tail"]: struct.append(f'this sentence cites no facts line: "{t[:80]}"') return dict(ok=not bad and not struct, sections=secs, notes=notes, bad=bad, numbers=nnum, names=nname, words=_words(render(secs)), sentences=ns, cited=sum(len(st["cites"]) for sec in secs for st in sec["sents"]), structure=struct, _F=F, _X=X) def _lead(t): """a paragraph whose opening sentence was dropped: its first clause takes the bold, so every paragraph still opens on one (the bold is the page's form, not a claim)""" if "**" in t: return t m = re.match(r"([^,;:*]{10,70})(?=[,;:]\s)", t) if m and 3 <= len(m.group(1).split()) <= 10 and m.group(1).count("`") % 2 == 0 and "[[" not in m.group(1): return f"**{m.group(1)}**" + t[m.end():] return t def render(secs): """the sections as the reading the page shows: no tags, no citations, one paragraph each""" return "\n\n".join(_lead(" ".join(st["text"] + st["punct"] for st in sec["sents"])) for sec in secs if sec["sents"]) def _words(t): return len(re.findall(r"[A-Za-z0-9][\w'.,%-]*", t.replace("**", ""))) def _carries(t): """what a sentence tells the reader: its numbers with their unit, its gene names, the columns it names — what is lost if it goes, and what another sentence already says if it stays""" out = ({" ".join(m.group(0).split()) for m in re.finditer(r"\d[\d,]*(?:\.\d+)?\s*(?:%|bits?|famil\w+)", t)} | {g.strip() for g in re.findall(r"`([^`\n]+)`", t)} | {f"c{c}" for c in cols_of(t)}) for m in re.finditer(r"\[\[(.+?)\]\]", t): out.add("[[" + m.group(1).lower() + "]]") return out def resolve(chk, limit=WORDS_MAX): """the reading to serve: the sentences whose facts hold, then sentences dropped until the text fits the word limit — the ones that add least (no id of their own, nothing of a sub-line or a branch, no interpretation) first, and never an opening sentence whose paragraph would be left with its promise alone. An opener that leant on its paragraph's other sentences is checked again against what is left of them: the citation holds for the text served, not only for the text written -> (text or None, dict(dropped, dropped_len, unled, sections, words, why))""" F, X = chk.get("_F"), chk.get("_X") secs = []; nbad = 0 for sec in chk["sections"]: keep = [st for st in sec["sents"] if st["ok"]] nbad += len(sec["sents"]) - len(keep) if keep: secs.append(dict(sec, sents=keep, n0=len(sec["sents"]))) # nothing of the section holds: the section goes rep = dict(dropped=nbad, dropped_len=0, unled=0, sections=[], words=0, repeats=0) seen_c = set() # a sentence whose numbers, genes and columns were for sec in secs: # all said before: a repeat, left out keep = [] for st in sec["sents"]: c = _carries(st["text"]) if c and c <= seen_c and len(sec["sents"]) > 1 and not (st is sec["sents"][0] and sec["tag"] in KEEP): rep["repeats"] += 1; continue seen_c |= c; keep.append(st) sec["sents"] = keep secs = [x for x in secs if x["sents"]] def prune(): """after every drop: an opener no longer carried by its own paragraph, and a paragraph left with nothing but the promise its opener made""" nonlocal secs for sec in secs: st = sec["sents"][0] if sec["sents"] else None if not (st and st["opens"] and st.get("inherit") and F is not None and X is not None): continue left = [i for s2 in sec["sents"][1:] for i in s2["cites"]] if all(i in left for i in st["inherit"]): continue b, _n1, _n2 = _sent_bad(st["text"], st["cites"], F, X, [i for i in st["inherit"] if i in left]) if b: sec["sents"] = sec["sents"][1:]; rep["dropped"] += 1 secs = [s for s in secs if s["sents"] and not (len(s["sents"]) == 1 and s["sents"][0]["opens"] and s["n0"] > 1 and not re.search(r"\d", s["sents"][0]["text"]) and s["tag"] not in KEEP)] prune() tags = [s["tag"] for s in secs] rep.update(sections=tags, words=_words(render(secs))) if any(t not in tags for t in KEEP) or len(secs) < len(SECTIONS) - 1: rep["why"] = "too little of the reading is supported by the facts it cites"; return None, rep for _ in range(40): # the word limit: whole sentences at a time if _words(render(secs)) <= limit: break all_ = [(s, j, st) for s in secs for j, st in enumerate(s["sents"])] carries = {(id(s), j): _carries(st["text"]) for s, j, st in all_} cand = [] for s, j, st in all_: if len(s["sents"]) == 1 and s["tag"] in KEEP: continue # @start and @whole stay others = set().union(set(), *(v for k, v in carries.items() if k != (id(s), j))) uniq = len(carries[(id(s), j)] - others) # the numbers, genes and columns it alone carries sub = any("." in i for i in st["cites"]); interp = "[[" in st["text"] if j != len(s["sents"]) - 1 and (uniq or sub or interp): continue # the tail of a paragraph, or a sentence if st["opens"] and len(s["sents"]) > 1: continue # that repeats what another one carries cand.append((uniq + 2 * sub + 2 * interp, DROP_ORDER.index(s["tag"]) if s["tag"] in DROP_ORDER else 9, -_words(st["text"]), id(s), s, j)) if not cand: break *_k, s, j = min(cand, key=lambda c: c[:4]) del s["sents"][j]; rep["dropped_len"] += 1 prune() text = render(secs) rep.update(words=_words(text), sections=[s["tag"] for s in secs], unled=sum(not s["sents"][0]["opens"] for s in secs), interp=len(re.findall(r"\[\[", text)), kept=[dict(tag=s["tag"], text=st["text"], cites=st["cites"], inherit=st.get("inherit") or []) for s in secs for st in s["sents"]]) cov = set() # the columns the reading served speaks of for s in secs: for st in s["sents"]: cov |= cols_of(st["text"]) | set().union(set(), *[X[i]["cols"] for i in st["cites"] if X and i in X and not X[i]["glob"]]) rep["covered"] = len(cov) if any(t not in rep["sections"] for t in KEEP) or len(secs) < len(SECTIONS) - 1: rep["why"] = "too little of the reading is supported by the facts it cites"; return None, rep if rep["words"] < WORDS_MIN: rep["why"] = f"only {rep['words']} words of the reading are supported by the facts"; return None, rep if rep["words"] > limit: rep["why"] = f"{rep['words']} words after the sections were trimmed"; return None, rep return text, rep def retry_cited(chk, rep, truncated=False): """what to tell the model about its rejected reading: the sentences whose facts are not in their citations, what the structure asks, and the length""" say = list(chk["structure"][:4]) n = 0 for sec in chk["sections"]: for st in sec["sents"]: if st["ok"] or n >= 6: continue n += 1 why = "; ".join(dict.fromkeys(f"{b['text']}: {b['why']}" for b in st["bad"][:3])) say.append(f'in @{sec["tag"]}, "{st["text"][:90]}" {{{", ".join(st["cites"]) or "no id"}}} — {why}') if truncated: say.append(f"the text was cut at its length limit: write at most {WORDS_MAX} words") elif chk["words"] > WORDS_TARGET: say.append(f"the reading is {chk['words']} words: write {WORDS_MIN_ASK} to {WORDS_TARGET}, " "so that nothing has to be cut") if rep and rep.get("unled"): say.append("a paragraph is left without its opening sentence: write an opener that stands on the " "lines its own paragraph cites") if rep and rep.get("interp", 1) == 0 and chk.get("_F") is not None and "Q" in chk["_F"].ids: say.append("the reading names no biology: put one [[interpretation]] of 2 to 6 words on a group of [Q], from the words [Q] " "gives it, naming that group's columns and citing its stretch line and [Q]") if rep and rep.get("why"): say.append(rep["why"]) return ("Your reading was rejected: " + "; ".join(say) + ". Rewrite the whole reading, following the form, the citations " "and the rules: every number, gene name and column of a sentence standing in the facts lines that sentence cites, " "nothing from any other line.") def reason(chk, truncated, rep=None): """a short account of a failed check, for the page (its details in `bad`)""" b = chk["bad"] if truncated and not b: return "the model's text was cut at its length limit" kinds = collections.Counter(x["kind"] for x in b) what = dict(number="a number the facts lines its sentence cites do not carry", scope="a number given to the wrong genes or columns", column="a gene or a column its sentence does not cite", direction="the comparison inside and outside regions of plasticity inverted", claim="a claim this window's data do not make", gene="a gene not in the lines its sentence cites", identifier="a name not in this window", interpretation="an interpretation the products and gene names do not carry", citation="a sentence written from facts it does not cite", wording="words about the facts rather than the window", format="a number badly written") what["number word"] = "a number in words" if not kinds and chk.get("structure"): return "the model's text did not keep the reading's sections after a second attempt" if not kinds and rep and rep.get("why"): return rep["why"] return "the model's text still had " + ", ".join(what.get(k, k) for k, _ in kinds.most_common(3)) + " after a second attempt" # ---------------------------------------------------------------- the reading as HTML (the page's classes) INTERP_TITLE = "interpretation from the gene names and products, not measured here" def to_html(text, F): text = re.sub(r"^\s*#+\s*", "", text.strip(), flags=re.M) # no headings genes = sorted(F.genes, key=len, reverse=True) gre = re.compile(r"(?\1", h) h = re.sub(r"(?{mm.group(1)}' if known(html.unescape(mm.group(1))) else mm.group(1), h) h = re.sub(r"\[\[(.+?)\]\]", rf'\1', h) h = h.replace("**", "").replace("`", "").replace("[[", "").replace("]]", "") if not balanced(h): h = html.escape(re.sub(r"\*\*|\[\[|\]\]|`", "", p), quote=True) out.append(f'

{h}

') return "".join(out) def balanced(h): st = [] for mm in re.finditer(r"<(/?)(b|span)\b[^>]*>", h): if not mm.group(1): st.append(mm.group(2)) elif not st or st.pop() != mm.group(2): return False return not st # ---------------------------------------------------------------- the LLM _TOKEN = [None, False] def token(): """HF_TOKEN, else the local login; read once, kept in memory only""" if not _TOKEN[1]: t = os.environ.get("HF_TOKEN") or None if not t: try: from huggingface_hub import get_token t = get_token() except Exception: t = None _TOKEN[0] = t; _TOKEN[1] = True return _TOKEN[0] class _NoRedirect(urllib.request.HTTPRedirectHandler): """a redirect is an error: the Authorization header is never sent to another address""" def redirect_request(self, *a, **k): return None OPENER = urllib.request.build_opener(_NoRedirect) def _bill(cost): with LOCK: _roll(); USAGE["cost"] += cost; USAGE["calls"] += 1 def llm(messages, deadline): """one completion: the providers in order, the next on a timeout, a redirect, a 429 or a 5xx. -> dict(text, finish, usage, provider, ms). Every call is counted in the day's dollars: a call cut by a timeout or without usage at its worst case (the prompt and MAX_TOKENS out). The router's error text is never passed on.""" tok = token() if not tok: raise Unavailable("no Hugging Face token on this server (HF_TOKEN)", code="no_token") errs = [] est_in = len(json.dumps(messages)) / 3.0 for prov in PROVIDERS: left = deadline - time.time() if left < 5: errs.append(f"{prov}: no time left"); break pin, pout = PRICES.get(prov, (0.3, 1.2)); worst = (est_in * pin + MAX_TOKENS * pout) / 1e6 body = dict(model=f"{MODEL}:{prov}", messages=messages, max_tokens=MAX_TOKENS, temperature=TEMPERATURE, chat_template_kwargs=dict(thinking=False), stream=False) req = urllib.request.Request(URL, data=json.dumps(body).encode(), method="POST", headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"}) t0 = time.time() try: with OPENER.open(req, timeout=min(TIMEOUT, left)) as r: out = json.load(r); used = r.headers.get("x-inference-provider") or prov except urllib.error.HTTPError as e: if e.code in (401, 403): AUTH.update(t=time.time(), code="refused", why=f"the token was refused by the Inference Providers router (HTTP {e.code})") raise Unavailable(AUTH["why"], code="refused") if e.code == 402: AUTH.update(t=time.time(), code="credit", why="no Inference Providers credit left on this token's account (HTTP 402)") raise Unavailable(AUTH["why"], code="credit") errs.append(f"{prov}: HTTP {e.code}"); continue except Exception as e: # timeout, connection: maybe billed _bill(worst); errs.append(f"{prov}: {type(e).__name__}"); continue try: ch = out["choices"][0]; text = (ch["message"].get("content") or "").strip() except Exception: _bill(worst); errs.append(f"{prov}: malformed answer"); continue u = out.get("usage") or {} pin, pout = PRICES.get(used, (pin, pout)) cost = u.get("estimated_cost") if cost is None and u.get("prompt_tokens") is not None: cost = (u.get("prompt_tokens", 0) * pin + (u.get("completion_tokens") or MAX_TOKENS) * pout) / 1e6 if cost is None: cost = worst _bill(float(cost)) if not text: errs.append(f"{prov}: empty answer"); continue usage = dict(prompt_tokens=u.get("prompt_tokens"), completion_tokens=u.get("completion_tokens"), cached_tokens=(u.get("prompt_tokens_details") or {}).get("cached_tokens"), reasoning_tokens=(u.get("completion_tokens_details") or {}).get("reasoning_tokens"), cost_usd=round(float(cost), 6)) return dict(text=text, finish=ch.get("finish_reason"), usage=usage, provider=used, ms=round((time.time() - t0) * 1000)) raise Upstream("the reading model could not be reached (" + "; ".join(errs) + ")", 30) def _attempt(r, F): """one answer of the model: tidied, parsed into its sections, every sentence checked against the lines it cites, then the reading that would be served (its unsupported sentences dropped, trimmed to the word limit)""" text = tidy(r["text"], F) chk = check_cited(text, F) out, rep = resolve(chk) cut = r["finish"] == "length" drops = [dict(tag=sec["tag"], text=st["text"][:200], cites=st["cites"], opens=st["opens"], bad=[{k: v for k, v in b.items() if k != "context"} for b in st["bad"][:4]]) for sec in chk["sections"] for st in sec["sents"] if not st["ok"]] return dict(r, raw=r["text"], text=out or text, annotated=text, check={k: v for k, v in chk.items() if k not in ("sections", "_F", "_X")}, report=rep, dropped=drops, served=out, truncated=cut, _chk=chk, ok=bool(out) and not cut and not chk["structure"] and rep["dropped"] <= 1 # one sentence dropped, five and len(rep["sections"]) == len(SECTIONS) and not rep["unled"] # sections left, every paragraph and (rep.get("interp", 0) >= 1 or "Q" not in F.ids)) # led, one interpretation: no 2nd call def generate(F, wid): """-> the cache entry: status llm (the sections that hold against their citations) or fallback (too little of the text was supported, twice)""" deadline = time.time() + BUDGET sysmsg, sha = system_msg() msgs = [dict(role="system", content=sysmsg), dict(role="user", content=user_msg(F, wid))] attempts = [] for i in range(2): try: r = llm(msgs, deadline) except Upstream: if not attempts: raise break # the retry could not be made: fallback a = _attempt(r, F); attempts.append(a) if a["ok"]: break # only its length was over: the trimming fits it if i == 0: if deadline - time.time() < 12: break msgs += [dict(role="assistant", content=r["text"]), dict(role="user", content=retry_cited(a["_chk"], a["report"], a["truncated"]))] clean = [a for a in attempts if a["ok"]] served = [a for a in attempts if a["served"] and not a["truncated"]] best = (clean[0] if clean else min(served, key=lambda a: (a["report"]["dropped"] + a["report"]["unled"], -a["report"].get("interp", 0), -a["report"].get("covered", 0))) if served else attempts[-1]) ok = bool(best["served"]) and not best["truncated"] usage = dict(prompt_tokens=sum(a["usage"]["prompt_tokens"] or 0 for a in attempts), completion_tokens=sum(a["usage"]["completion_tokens"] or 0 for a in attempts), cost_usd=round(sum(a["usage"]["cost_usd"] for a in attempts), 6), ms=sum(a["ms"] for a in attempts)) c = best["check"]; rep = best["report"] e = dict(status="llm" if ok else "fallback", model=MODEL, provider=best["provider"], created=time.time(), usage=usage, attempts=[{k: v for k, v in a.items() if k != "_chk"} for a in attempts], checked=dict(ok=ok, numbers=c["numbers"], names=c["names"], words=rep["words"] if ok else c["words"], attempts=len(attempts), sentences=c["sentences"], cited=c["cited"], skill=sha, dropped=rep["dropped"], trimmed=rep["dropped_len"], repeats=rep.get("repeats", 0), unled=rep["unled"], sections=rep["sections"], interp=rep.get("interp", 0), covered=rep.get("covered", 0))) if ok: e.update(text=best["served"], html=to_html(best["served"], F)) else: e["reason"] = reason(c, best["truncated"], rep) e["checked"]["bad"] = [{k: v for k, v in b.items() if k != "context"} for b in c["bad"][:12]] e["checked"]["structure"] = c["structure"][:4] return e # ---------------------------------------------------------------- cache, limits, the route LOCK = threading.Lock() MEM = collections.OrderedDict() # window -> entry (with its facts hash) FACTS = collections.OrderedDict() # window -> (Facts, hash) INFLIGHT = {} # window -> dict(ev, waiters, err): the generation under way SLOTS = threading.BoundedSemaphore(MAX_PARALLEL) QUEUED = [0] # new generations waiting for a slot HITS = {} # client -> deque of its new generations' times (PER_IP_S) DAYHITS = {} # client -> its new generations today PRUNED = [0.0] AUTH = dict(t=0.0, why=None, code=None) USAGE = dict(day=None, n=0, cost=0.0, calls=0) _HOME = [None] def _safe(wid): return re.sub(r"[^A-Za-z0-9_.-]", "_", wid)[:80] def _paths(wid): ds = [CACHE_DIR] + ([BUNDLED] if os.path.abspath(BUNDLED) != os.path.abspath(CACHE_DIR) else []) return [os.path.join(d, _safe(wid) + ".json") for d in ds] def _today(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d") def _to_midnight(): n = datetime.datetime.now(datetime.timezone.utc) return int((n.replace(hour=0, minute=0, second=0, microsecond=0) + datetime.timedelta(days=1) - n).total_seconds()) + 1 def _roll(): """a new UTC day: the day's counters start again (under LOCK)""" if USAGE["day"] != _today(): USAGE.update(day=_today(), n=0, cost=0.0, calls=0); DAYHITS.clear() def _usage_load(): try: u = json.load(open(os.path.join(CACHE_DIR, "_usage.json"))) if u.get("day") == _today(): USAGE.update(day=u["day"], n=int(u["n"]), cost=float(u.get("cost", 0)), calls=int(u.get("calls", 0))) except Exception: pass _roll() def _write(path, obj): try: os.makedirs(os.path.dirname(path), exist_ok=True); tmp = f"{path}.{os.getpid()}.{threading.get_ident()}.tmp" with open(tmp, "w") as f: json.dump(obj, f, indent=1) os.replace(tmp, path) except OSError: pass # a read-only disk: memory only def home_data(): if _HOME[0] is None: _HOME[0] = json.load(open("pgb/graph_pgb.json")) return _HOME[0] def window_id(a): w = (a.get("window") or a.get("anchor") or "").strip() if not w: raise ValueError("give window=dnaA or anchor=") if w.lower() == DNAA.lower(): return DNAA if not re.fullmatch(r"[A-Za-z0-9_.-]{1,64}", w): raise ValueError("unknown window") return w def facts_of(wid, region): with LOCK: if wid in FACTS: FACTS.move_to_end(wid); return FACTS[wid] D = home_data() if wid == DNAA else region(wid) F = facts(D, wid); sysmsg, sha = system_msg() # the skill's sha is part of the key: a new skill, a new reading h = hashlib.sha1((PROMPT_VERSION + MODEL + sha + sysmsg + user_msg(F, wid)).encode()).hexdigest()[:12] F.label = DNAA if wid == DNAA else D["meta"].get("anchor_label") or wid with LOCK: FACTS[wid] = (F, h) while len(FACTS) > 64: FACTS.popitem(last=False) return F, h def cached(wid, h): with LOCK: e = MEM.get(wid) if e is not None: MEM.move_to_end(wid) if e is None or e.get("facts_hash") != h: e = None for p in _paths(wid): try: x = json.load(open(p)) except Exception: continue if x.get("facts_hash") == h: e = x; break if e is None: return None if e["status"] == "fallback" and time.time() - e["created"] > FAIL_TTL: return None with LOCK: MEM[wid] = e while len(MEM) > 256: MEM.popitem(last=False) return e def availability(): """-> None when new readings can be written here, else (code, why)""" if os.environ.get("READING", "").lower() in ("off", "0", "no", "false") or PER_IP <= 0 or PER_IP_DAY <= 0: return "off", "new readings are turned off on this server (READING=off)" if not token(): return "no_token", "no Hugging Face token on this server (set the HF_TOKEN secret)" if AUTH["why"] and time.time() - AUTH["t"] < AUTH_RETRY: return AUTH["code"], AUTH["why"] _roll() if USAGE["n"] >= DAILY_CAP: return "cap", f"the {DAILY_CAP} new readings of the day are written (until 00:00 UTC)" if USAGE["cost"] >= DAILY_USD: return "spend", f"the day's ${DAILY_USD:g} for new readings is spent (until 00:00 UTC)" return None def _ip(x): """an X-Forwarded-For entry as an address: brackets and a port stripped; None if it is not one""" x = x.strip().strip('"') m = re.fullmatch(r"\[([0-9A-Fa-f:.]+)\](?::\d+)?", x) if m: x = m.group(1) elif re.fullmatch(r"[\d.]+:\d+", x): x = x.rsplit(":", 1)[0] try: return ipaddress.ip_address(x) except ValueError: return None def client_ip(peer, xff): """the client, for its limits. Behind a trusted proxy (a Hugging Face Space, SPACE_ID set, or TRUST_PROXY=1): the rightmost public address of X-Forwarded-For (all its lines joined), the one the proxy appended; the walk stops at an entry that is not an address, which is then the key (a client can only prepend). Else the peer. IPv6 by /64.""" key = None if TRUST_PROXY and xff: for x in reversed([s for s in xff.split(",") if s.strip()]): ip = _ip(x) if ip is None: key = "?" + x.strip()[:64]; break if ip.is_private or ip.is_loopback or ip.is_link_local: continue key = ip; break if key is None: key = _ip(peer or "") or (peer or "?") if isinstance(key, ipaddress.IPv6Address) and not key.ipv4_mapped: return str(ipaddress.ip_network(f"{key}/64", strict=False)) if isinstance(key, ipaddress.IPv6Address): return str(key.ipv4_mapped) return str(key) def site_ok(sfs, origin, host): """a new reading only for the page's own site: Sec-Fetch-Site same-origin (or none: typed in), an Origin of this host; locally (not a Space) also a page on localhost or opened as a file (Origin null). A request with no browser metadata (a script) passes: it has the same limits. Cached readings (peek, or written already) go to anyone.""" sfs = (sfs or "").strip().lower(); origin = (origin or "").strip() if sfs in ("same-origin", "none"): return True if origin: if origin == "null": return not SPACE o = urllib.parse.urlparse(origin).netloc.lower() if host and o == host.strip().lower(): return True return not SPACE and bool(re.fullmatch(r"(localhost|127\.0\.0\.1|\[::1\])(:\d+)?", o)) return sfs not in ("cross-site", "same-site") def _prune(now): """forget the clients with no new reading in the last PER_IP_S (under LOCK)""" if now - PRUNED[0] < 60: return PRUNED[0] = now for k in [k for k, q in HITS.items() if not q or now - q[-1] > PER_IP_S]: del HITS[k] def _admit(ip, commit): """may this client have a new generation now? commit: count it, for the client and the day""" now = time.time() with LOCK: why = availability() if why: raise Unavailable(why[1], _to_midnight() if why[0] in ("cap", "spend") else None, why[0]) _prune(now) q = HITS.get(ip) while q and now - q[0] > PER_IP_S: q.popleft() if q and len(q) >= PER_IP: raise Limited(f"{PER_IP} new readings per {PER_IP_S / 60:.0f} minutes from one visitor", int(PER_IP_S - (now - q[0])) + 1, "limit") d = DAYHITS.get(ip, 0) if d >= PER_IP_DAY: raise Limited(f"{PER_IP_DAY} new readings a day from one visitor", _to_midnight(), "limit_day") if commit: HITS.setdefault(ip, collections.deque()).append(now); DAYHITS[ip] = d + 1; USAGE["n"] += 1 def _refund(ip): """a generation that could not be made: the client may ask again (the day's count and dollars stay)""" with LOCK: q = HITS.get(ip) if q: q.pop() if DAYHITS.get(ip): DAYHITS[ip] -= 1 def _slot(): """one of the MAX_PARALLEL generation slots: a free one at once, else wait (at most QUEUE_MAX wait, SLOT_WAIT each)""" if SLOTS.acquire(blocking=False): return with LOCK: if QUEUED[0] >= QUEUE_MAX: raise Busy("other readings are being written", 10) QUEUED[0] += 1 try: ok = SLOTS.acquire(timeout=SLOT_WAIT) finally: with LOCK: QUEUED[0] -= 1 if not ok: raise Busy("other readings are being written", 5) def public(e, wid, F, cached_, with_facts=False): out = dict(window=wid, label=F.label, status=e["status"], model=e["model"], provider=e.get("provider"), cached=cached_, created=datetime.datetime.fromtimestamp(e["created"], datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), checked=e["checked"], usage=dict(e["usage"])) if e["status"] == "llm": out.update(html=e["html"], text=e["text"]) else: out["reason"] = e.get("reason", "") if with_facts: out["facts"] = F.txt() return out def answer(a, peer="", xff="", region=None, site=True): """-> (HTTP code, JSON) for GET /reading?window=dnaA | anchor= [&peek=1] [&facts=1]; raises a ReadingError (its status and body()) or ValueError / KeyError (400). site: the request comes from the page's own site""" t0 = time.time(); wid = window_id(a) try: F, h = facts_of(wid, region) except KeyError as e: raise ValueError(e.args[0] if e.args else "unknown window") wf = a.get("facts") == "1" done = lambda e, c: (200, dict(public(e, wid, F, c, wf), ms=round((time.time() - t0) * 1000))) e = cached(wid, h) if e: return done(e, True) if a.get("peek") == "1": with LOCK: why = availability() out = dict(window=wid, label=F.label, status="none", model=MODEL, available=why is None) if why: out.update(code=why[0], reason=why[1]) if wf: out["facts"] = F.txt() return 200, out if not site: raise Refused("new readings are written only for the page of this server") ip = client_ip(peer, xff) with LOCK: u0 = (USAGE["n"], USAGE["calls"]) for _ in range(3): # one generation per window at a time with LOCK: job = INFLIGHT.get(wid); mine = job is None if mine: job = INFLIGHT[wid] = dict(ev=threading.Event(), waiters=0, err=None) elif job["waiters"] >= WAITERS: raise Busy("this window's reading is being written", 5) else: job["waiters"] += 1 if mine: break try: job["ev"].wait(BUDGET + SLOT_WAIT + 5) finally: with LOCK: job["waiters"] -= 1 e = cached(wid, h) if e: return done(e, True) err = job["err"] if isinstance(err, (Unavailable, Upstream)): raise type(err)(str(err), err.retry_s, err.code) # the first request was refused for itself (its limit, its site) or had to wait: this one tries on its own else: raise Busy("this window's reading is being written", 5) try: _admit(ip, False) _slot() try: _admit(ip, True) try: e = generate(F, wid) except (Upstream, Unavailable): _refund(ip); raise finally: SLOTS.release() e.update(window=wid, label=F.label, facts_hash=h, prompt_version=PROMPT_VERSION, skill_sha=skill()[1], facts=F.txt()) with LOCK: MEM[wid] = e while len(MEM) > 256: MEM.popitem(last=False) _write(_paths(wid)[0], e) return done(e, False) except ReadingError as x: job["err"] = x; raise finally: with LOCK: INFLIGHT.pop(wid, None); u = dict(USAGE) job["ev"].set() if (u["n"], u["calls"]) != u0: _write(os.path.join(CACHE_DIR, "_usage.json"), u) def health(): with LOCK: why = availability(); n = len(MEM); clients = len(HITS); u = dict(USAGE) nf = set() for d in {CACHE_DIR, BUNDLED}: try: nf |= {f for f in os.listdir(d) if f.endswith(".json") and not f.startswith("_")} except OSError: pass out = dict(available=why is None, model=MODEL, providers=PROVIDERS, max_tokens=MAX_TOKENS, cached=max(n, len(nf)), prompt=dict(version=PROMPT_VERSION, skill=skill()[1], skill_file=_SKILL[2], temperature=TEMPERATURE, words=WORDS_MAX), today=dict(new=u["n"], cap=DAILY_CAP, calls=u["calls"], cost_usd=round(u["cost"], 4), usd_cap=DAILY_USD), per_client=dict(new=PER_IP, per_s=int(PER_IP_S), per_day=PER_IP_DAY, clients=clients, proxy=TRUST_PROXY)) if why: out.update(reason=why[1], code=why[0]) return out _usage_load() def pregen(args): """write the readings of windows ahead, without the server (python3 pg_reading.py pregen ...)""" import concurrent.futures as cf import pgb_region as RG dry = "--dry" in args; ws = [a for a in args if not a.startswith("--")] if "--chain" in args: # the page's Next from the dnaA window, then Next again; Previous from each a = RG.resolve("60"); seen = [] while a is not None and a not in seen: seen.append(a); _, nx = RG.neighbours(a, 60); a = RG.FIDX[nx["fam"]] if nx else None prev = [RG.FIDX[p["fam"]] for p in (RG.neighbours(a, 60)[0] for a in seen) if p and p["pos"] != 0] ws += [RG.FN[a] for a in seen] + [RG.FN[a] for a in prev if a not in seen] ws = list(dict.fromkeys(ws)) region = lambda w: RG.build(RG.FIDX[w]) todo = [] for w in ws: wid = DNAA if w.lower() == DNAA.lower() else w if wid != DNAA and RG.FIDX.get(wid) not in RG.BB_RANK: print(f"{w}: not a window anchor", flush=True); continue F, h = facts_of(wid, region) if cached(wid, h): continue todo.append((wid, F, h)) print(f"{len(ws)} windows, {len(todo)} to write", flush=True) if dry: return tot = dict(cost=0.0, n=0, fb=0) def one(x): wid, F, h = x; t = time.time() try: e = generate(F, wid) except ReadingError as err: return wid, f"error {err.code}: {err}", None e.update(window=wid, label=F.label, facts_hash=h, prompt_version=PROMPT_VERSION, skill_sha=skill()[1], facts=F.txt()) _write(_paths(wid)[0], e) return wid, f"{e['status']} {e['checked']['attempts']} attempt(s) ${e['usage']['cost_usd']:.5f} {time.time() - t:.1f}s", e with cf.ThreadPoolExecutor(MAX_PARALLEL) as ex: for wid, msg, e in ex.map(one, todo): if e: tot["cost"] += e["usage"]["cost_usd"]; tot["n"] += 1; tot["fb"] += e["status"] == "fallback" print(wid, msg, flush=True) print(f"written {tot['n']} ({tot['fb']} fallback), ${tot['cost']:.4f}; calls counted ${USAGE['cost']:.4f}", flush=True) if __name__ == "__main__": import sys if len(sys.argv) >= 3 and sys.argv[1] == "facts": w = sys.argv[2]; D = home_data() if w == DNAA else json.load(open(w)) F = facts(D, DNAA if w == DNAA else D["meta"]["anchor"]); t = F.txt() print(t); print(f"\n[{len(t)} characters, ~{len(t) // 4} tokens; {sum(map(len, F.nums.values()))} numbers, {len(F.genes)} genes]", file=sys.stderr) elif len(sys.argv) >= 4 and sys.argv[1] == "check": w = sys.argv[2]; D = home_data() if w == DNAA else json.load(open(w)) F = facts(D, DNAA if w == DNAA else D["meta"]["anchor"]) chk = check_cited(tidy(open(sys.argv[3]).read(), F), F); out, rep = resolve(chk) for sec in chk["sections"]: for st in sec["sents"]: print(f"{'ok ' if st['ok'] else 'DROP'} @{sec['tag']:8s} {{{', '.join(st['cites'])}}} {st['text'][:110]}") for b in st["bad"]: print(f" - {b['kind']}: {b['text']} — {b['why']}") print(json.dumps(dict({k: v for k, v in chk.items() if k not in ("sections", "bad", "notes")}, report=rep), indent=1)) old = check(render(chk["sections"]), F) # the whole-text check of r7, for comparison print(f"the whole-text check: {'ok' if old['ok'] else 'not ok'}, {len(old['bad'])} item(s) " + ", ".join(sorted({b['kind'] for b in old['bad']}))) print("\n" + (out or "(nothing servable)")) elif len(sys.argv) >= 2 and sys.argv[1] == "skill": t, sha = skill(); print(f"[{SKILL_PATH} {'read' if _SKILL[2] else 'MISSING'}, sha {sha}, {len(t)} characters, ~{len(t) // 4} tokens]") sysmsg, _ = system_msg(); print(f"[system message {len(sysmsg)} characters, ~{len(sysmsg) // 4} tokens]") elif len(sys.argv) >= 2 and sys.argv[1] == "pregen": pregen(sys.argv[2:]) else: print(__doc__)