Spaces:
Running on Zero
Running on Zero
kwabs22
Claude Opus 5.5
Add a pronoun rule option: most recent object or last clause's subject
6287002 | """Ties the stages together: ledger -> settled meshes -> configuration -> files. | |
| Refining happens on the serial side: frozen items keep their decisions, | |
| everything else may change. Configuration is always free (seeded). | |
| """ | |
| import json | |
| import os | |
| import tempfile | |
| from . import assets, html, scene | |
| from .grammar import (POOL, build_ledger, candidate_label, extend_ledger, next_tier, parse_primitives, | |
| primitive_candidate) | |
| from .lexicon import ASSET_TIERS, LEXICON | |
| def _workdir(): | |
| return tempfile.mkdtemp(prefix="vli_") | |
| def _option_label(cand): | |
| """Objaverse candidates share a category name, so the uid tells them apart.""" | |
| label = candidate_label(cand) | |
| return f"{label} [{cand['uid'][:6]}]" if cand.get("uid") else label | |
| def _on_reject(item): | |
| """What rejecting this item's tier would offer: alternatives in the next asset tier | |
| are invisible in the question until then.""" | |
| dec = item["decision"] | |
| if dec["tier"] not in ASSET_TIERS: | |
| return None | |
| found = LEXICON.lookup(item["word"]) | |
| for tier in ASSET_TIERS[ASSET_TIERS.index(dec["tier"]) + 1:]: | |
| n = len(found.get(tier, [])) | |
| if n: | |
| return {"tier": tier, "candidates": n, "offered": min(n, POOL)} | |
| return {"tier": "primitive"} | |
| def settle_all(ledger): | |
| """Load every object's mesh; unusable candidates are skipped and recorded.""" | |
| meshes = {} | |
| for it in ledger["items"]: | |
| if it["kind"] == "object": | |
| full = assets.settle(it) | |
| meshes[it["id"]] = assets.scene_mesh(full) | |
| it["decision"]["label"] = candidate_label(it["decision"]["chosen"]) | |
| for q in ledger["unresolved"]: | |
| if q.get("binds"): | |
| continue # a reference: which item a word means, not which asset | |
| about = next((i for i in ledger["items"] if i["id"] == q.get("about")), None) | |
| if about and about["kind"] == "object": | |
| q["default_chosen"] = _option_label(about["decision"]["chosen"]) | |
| q["alternatives"] = [_option_label(c) for c in about["decision"]["alternatives"]] | |
| q["on_reject"] = _on_reject(about) | |
| return meshes | |
| def outputs(ledger, seed): | |
| meshes = settle_all(ledger) | |
| config = scene.configure(ledger, meshes, seed) | |
| ledger["configuration"] = config | |
| work = _workdir() | |
| png = scene.render_scene(ledger, meshes, config, os.path.join(work, "scene.png")) | |
| glb = scene.export_glb(ledger, meshes, config, os.path.join(work, "scene.glb")) | |
| return json.dumps(ledger, indent=1), png, glb, html.scene_page(ledger, meshes, config) | |
| def translate(sentence, primitives=None, seed=0, pronouns="recent"): | |
| return outputs(build_ledger(sentence, primitives, pronouns=pronouns), seed) | |
| def extend(ledger, sentence, primitives=None, seed=0, pronouns=None): | |
| return outputs(extend_ledger(ledger, sentence, primitives, pronouns), seed) | |
| def _bind(ledger, src, dst): | |
| """Two words mean one thing. `src` is a duplicate object (merged into `dst`) or a | |
| reference question (its word now means `dst`). Recorded in ledger["resolutions"].""" | |
| by_id = {i["id"]: i for i in ledger["items"]} | |
| target = by_id.get(dst) | |
| if target is None or target["kind"] != "object": | |
| raise ValueError(f"bind target {dst!r} is not an object item") | |
| relations = [i for i in ledger["items"] if i["kind"] == "relation"] | |
| q = next((q for q in ledger["unresolved"] if q["id"] == src and q.get("binds")), None) | |
| if q is not None: | |
| tok = q["binds"]["token"] | |
| moves = [(r, role) for r in relations for role in ("subject", "object") | |
| if r.get("tokens", {}).get(role) == tok] | |
| word, old = ledger["tokens"][tok]["text"] if tok < len(ledger["tokens"]) else "?", q["binds"]["to"] | |
| else: | |
| it = by_id.get(src) | |
| if it is None or it["kind"] != "object": | |
| raise ValueError(f"{src!r} is neither an object item nor a reference question") | |
| if it is target: | |
| raise ValueError(f"{src} is already {dst}") | |
| if it["frozen"]: | |
| raise ValueError(f"{src} is frozen; unfreeze it first") | |
| moves = [(r, role) for r in relations for role in ("subject", "object") if r[role] == src] | |
| word, old = it["word"], src | |
| for r, role in moves: # check before changing anything | |
| if r["object" if role == "subject" else "subject"] == dst: | |
| raise ValueError(f"binding {src} to {dst} would relate the {target['word']} to itself ({r['id']})") | |
| for r, role in moves: | |
| r[role] = dst | |
| if q is not None: | |
| ledger["tokens"][tok]["refers_to"] = dst | |
| ledger["unresolved"].remove(q) | |
| else: | |
| ledger["items"].remove(it) | |
| if it.get("emits_light"): | |
| target["emits_light"] = True | |
| for t in ledger["tokens"]: | |
| if src in t["commits"]: | |
| t["commits"].remove(src) | |
| t["refers_to"] = dst | |
| elif t.get("refers_to") == src: | |
| t["refers_to"] = dst | |
| kept = [] | |
| for u in ledger["unresolved"]: | |
| if u.get("about") == src and not u.get("binds"): | |
| continue # "which cat?" for the duplicate: the duplicate is gone | |
| if u.get("binds", {}).get("to") == src: | |
| u["binds"]["to"], u["about"], u["default_chosen"] = dst, dst, f"the {target['word']} ({dst})" | |
| kept.append(u) | |
| ledger["unresolved"] = kept | |
| live = {u["id"] for u in ledger["unresolved"]} | |
| for t in ledger["tokens"]: # answered questions leave the reading too; the resolution note keeps them | |
| t["opens"] = [o for o in t.get("opens", []) if o in live] | |
| ledger.setdefault("resolutions", []).append( | |
| {"word": word, "from": old, "to": dst, "note": f"{old} -> {dst} (same referent)"}) | |
| def apply_changes(ledger, changes): | |
| """changes: list of | |
| {"id": ..., "freeze": true|false} | |
| {"id": ..., "use": <index into alternatives>} | |
| {"id": ..., "reject": true} -> next tier | |
| {"id": ..., "height_m": 0.4} | |
| {"id": ..., "shape": "sphere"} -> primitive tier with that shape | |
| {"reroll": true} -> every unfrozen object takes its next alternative | |
| {"id": "n12", "bind": "n1"} -> n12 is the same thing as n1: merged into it | |
| {"id": "q11", "bind": "n5"} -> the word behind reference question q11 means n5 | |
| Reroll walks the pool (the old choice goes to the back), so repeated rerolls | |
| visit every alternative before repeating, independent of the seed. | |
| """ | |
| by_id = {i["id"]: i for i in ledger["items"]} | |
| for ch in changes: | |
| if "bind" in ch: | |
| _bind(ledger, ch.get("id"), ch["bind"]) | |
| by_id = {i["id"]: i for i in ledger["items"]} | |
| continue | |
| if ch.get("reroll"): | |
| for it in ledger["items"]: | |
| dec = it.get("decision") | |
| if it["kind"] == "object" and not it["frozen"] and dec["alternatives"]: | |
| dec["alternatives"].append(dec["chosen"]) | |
| dec["chosen"] = dec["alternatives"].pop(0) | |
| continue | |
| it = by_id.get(ch.get("id")) | |
| if it is None: | |
| raise ValueError(f"no item with id {ch.get('id')!r}") | |
| if "freeze" in ch: | |
| it["frozen"] = bool(ch["freeze"]) | |
| continue | |
| if it["frozen"]: | |
| raise ValueError(f"{it['id']} is frozen; unfreeze it first") | |
| if it["kind"] != "object": | |
| raise ValueError(f"{it['id']} is a {it['kind']}; only object decisions can change") | |
| dec = it["decision"] | |
| if "use" in ch: | |
| k = int(ch["use"]) | |
| dec["alternatives"][k], dec["chosen"] = dec["chosen"], dec["alternatives"][k] | |
| if ch.get("reject"): | |
| nxt = next_tier(dec, it["word"]) | |
| dec["rejected_tiers"].append(dec["tier"]) | |
| dec.update(nxt) | |
| if it["height_m"]["source"] != "user": | |
| it["height_m"] = {"value": None, "source": "pending"} | |
| if "shape" in ch: | |
| shape = parse_primitives({it["word"]: ch["shape"]})[it["word"]] | |
| if dec["tier"] != "primitive": | |
| dec["rejected_tiers"].append(dec["tier"]) | |
| dec.update(tier="primitive", chosen=primitive_candidate(shape), alternatives=[]) | |
| if "height_m" in ch: | |
| it["height_m"] = {"value": float(ch["height_m"]), "source": "user"} | |
| return ledger | |