API reference

Every public function and the Theory class, in the groups the R package's reference index uses and in the same order, so a name is found in the same place on either site. The R package prefixes each name with tf_, so check() here is tf_check() there.

The four functions whose name matches their module (diagram, dossier, implications, simulate) are documented by their canonical path, so that the function rather than the module is shown.

Core IO

Read, validate and write theory objects. Theory is also the builder: adding a construct, a proposition or a prediction, and logging the provenance of each, are methods on the class here, where the R package exposes them as tf_add_* functions.

theoryforge.Theory

A theory as a versioned, machine-checkable object.

Wraps the parsed mapping (self.data); all accessors tolerate missing optional collections by treating them as empty.

Source code in src/theoryforge/core.py

class Theory:
    """A theory as a versioned, machine-checkable object.

    Wraps the parsed mapping (``self.data``); all accessors tolerate missing
    optional collections by treating them as empty.
    """

    def __init__(self, data: dict):
        if not isinstance(data, dict):
            raise TypeError("Theory data must be a mapping")
        self.data = data

    # -- convenience accessors -------------------------------------------------
    @property
    def id(self) -> str:
        return self.data.get("id", "")

    @property
    def maturity(self) -> str:
        return self.data.get("maturity", "")

    def _list(self, key: str) -> list:
        v = self.data.get(key)
        return v if isinstance(v, list) else []

    # -- validation ------------------------------------------------------------
    def validate(self, *, full: bool = False) -> bool:
        """Structural validation against the schema's required fields and enums.

        Returns True on success, raises ValueError listing every problem found.
        With ``full=True`` additionally checks referential integrity: that every
        id is unique within its collection and that every cross-reference
        (proposition endpoints, prediction derivations and diagnostics,
        assumption/evidence/test-outcome targets) points to a declared id, and
        that every prediction ``severity`` is a number within [0, 1]. The
        ``full`` checks are deterministic.
        """
        errors: list[str] = []
        d = self.data
        for req in ("schema_version", "id", "title", "maturity"):
            if not _nonempty_str(d.get(req)):
                errors.append(f"missing/empty required field: {req}")
        # Each enum test asks whether the value is a nonempty string before
        # asking whether it is a member. A YAML list or mapping reaches these
        # lines whenever a field is mistyped, and `x in <set>` raises TypeError
        # on an unhashable value, which would abandon the collected errors and
        # report something the contract never promised. The string test also
        # keeps the twins level: R's `%in%` coerces a one-element list to its
        # element, so `theory_form: [network]` used to validate there and be
        # refused here.
        mat = d.get("maturity")
        if not (_nonempty_str(mat) and mat in _MATURITY):
            errors.append(f"maturity must be one of {', '.join(sorted(_MATURITY))}")
        if "theory_form" in d:
            form = d["theory_form"]
            if not (_nonempty_str(form) and form in _FORM):
                errors.append(f"theory_form must be one of {', '.join(sorted(_FORM))}")
        # A misspelt collection key would otherwise pass silently and quietly
        # change every downstream verdict (a renamed `predictions:` drops the
        # whole collection), so unrecognised top-level fields are refused.
        known = _resources.theory_schema().get("properties", {})
        for key in d:
            if key not in known:
                errors.append(f"unknown top-level field: {key}")
        for i, c in enumerate(self._list("constructs")):
            for req in ("id", "label", "definition"):
                if not _nonempty_str(_field(c, req)):
                    errors.append(f"construct[{i}] missing/empty {req}")
        for i, p in enumerate(self._list("propositions")):
            for req in ("id", "from", "to", "relation"):
                if not _nonempty_str(_field(p, req)):
                    errors.append(f"proposition[{i}] missing/empty {req}")
            rel = _field(p, "relation")
            if _nonempty_str(rel) and rel not in _RELATION:
                errors.append(f"proposition[{i}] relation '{rel}' not allowed")
        for i, p in enumerate(self._list("predictions")):
            for req in ("id", "statement", "type"):
                if not _nonempty_str(_field(p, req)):
                    errors.append(f"prediction[{i}] missing/empty {req}")
            ty = _field(p, "type")
            if _nonempty_str(ty) and ty not in _PRED_TYPE:
                errors.append(f"prediction[{i}] type '{ty}' not allowed")

        if full:
            self._referential_errors(errors)

        if errors:
            raise ValueError("invalid theory object: " + "; ".join(errors))
        return True

    def _referential_errors(self, errors: list[str]) -> None:
        """Append referential-integrity problems (used by ``validate(full=True)``).

        Deterministic and mirrored byte-for-byte by the R implementation: the
        same checks in the same order with the same message text.
        """
        cons = self._list("constructs")
        props = self._list("propositions")
        preds = self._list("predictions")
        alts = self._list("alternatives")
        auxs = self._list("auxiliary_assumptions")

        def ids_of(items: list) -> set:
            return {i for i in (_field(it, "id") for it in items) if _nonempty_str(i)}

        construct_ids = ids_of(cons)
        proposition_ids = ids_of(props)
        prediction_ids = ids_of(preds)
        alternative_ids = ids_of(alts)

        def dups(items: list, kind: str) -> None:
            seen: set = set()
            for it in items:
                i = _field(it, "id")
                if _nonempty_str(i):
                    if i in seen:
                        errors.append(f"duplicate {kind} id: {i}")
                    seen.add(i)

        dups(cons, "construct")
        dups(props, "proposition")
        dups(preds, "prediction")
        dups(alts, "alternative")
        dups(auxs, "assumption")
        for i, p in enumerate(props):
            frm, to = _field(p, "from"), _field(p, "to")
            if _nonempty_str(frm) and frm not in construct_ids:
                errors.append(f"proposition[{i}] from '{frm}' is not a known construct")
            if _nonempty_str(to) and to not in construct_ids:
                errors.append(f"proposition[{i}] to '{to}' is not a known construct")
        for i, p in enumerate(preds):
            for dref in _as_list(_field(p, "derives_from")):
                if _nonempty_str(dref) and dref not in proposition_ids:
                    errors.append(f"prediction[{i}] derives_from '{dref}' is not a known proposition")
            for dv in _as_list(_field(p, "diagnostic_vs")):
                if _nonempty_str(dv) and dv not in alternative_ids:
                    errors.append(f"prediction[{i}] diagnostic_vs '{dv}' is not a known alternative")
        for i, a in enumerate(auxs):
            for pr in _as_list(_field(a, "protects")):
                if _nonempty_str(pr) and pr not in prediction_ids:
                    errors.append(f"assumption[{i}] protects '{pr}' is not a known prediction")
        for i, t in enumerate(self._list("test_outcomes")):
            pid = _field(t, "prediction_id")
            if _nonempty_str(pid) and pid not in prediction_ids:
                errors.append(f"test_outcome[{i}] prediction_id '{pid}' is not a known prediction")
        for i, e in enumerate(self._list("evidence")):
            s = _field(e, "supports")
            if _nonempty_str(s) and s not in prediction_ids:
                errors.append(f"evidence[{i}] supports '{s}' is not a known prediction")
        # The schema types prediction severity as a number in [0, 1]; enforced
        # here so a file cannot pass full validation and then be refused by
        # the scorer, which rejects non-numeric severities.
        for i, p in enumerate(preds):
            s = _field(p, "severity")
            if s is None:
                continue
            if (isinstance(s, bool) or not isinstance(s, (int, float))
                    or math.isnan(s) or s < 0 or s > 1):
                errors.append(f"prediction[{i}] severity must be a number between 0 and 1")

    # -- serialisation ---------------------------------------------------------
    def write(self, path) -> None:
        path = Path(path)
        if path.suffix.lower() == ".json":
            text = json.dumps(self.data, indent=2, ensure_ascii=False)
        else:
            text = yaml.safe_dump(self.data, sort_keys=False, allow_unicode=True)
        _write_lf(path, text)

    # -- builder (BUILDING mode) ----------------------------------------------
    def _provenance(self, action: str, detail: str) -> None:
        prov = self.data.setdefault("provenance", [])
        prov.append({"step": str(len(prov) + 1), "action": action, "detail": detail})

    def _coll(self, key: str) -> list:
        return self.data.setdefault(key, [])

    def add_construct(self, id, label, definition, measurement=None, boundary_conditions=None):
        c = {"id": id, "label": label, "definition": definition}
        if measurement is not None:
            c["measurement"] = list(measurement)
        if boundary_conditions is not None:
            c["boundary_conditions"] = list(boundary_conditions)
        self._coll("constructs").append(c)
        self._provenance("tf_add_construct", id)
        return self

    def add_proposition(self, id, frm, to, relation, mechanism=None):
        p = {"id": id, "from": frm, "to": to, "relation": relation}
        if mechanism is not None:
            p["mechanism"] = mechanism
        self._coll("propositions").append(p)
        self._provenance("tf_add_proposition", id)
        return self

    def add_prediction(self, id, statement, type, derives_from=None, diagnostic_vs=None):
        p = {"id": id, "statement": statement, "type": type}
        if derives_from is not None:
            p["derives_from"] = list(derives_from)
        if diagnostic_vs is not None:
            p["diagnostic_vs"] = list(diagnostic_vs)
        self._coll("predictions").append(p)
        self._provenance("tf_add_prediction", id)
        return self

    def add_alternative(self, id, label, key_constructs=None):
        a = {"id": id, "label": label}
        if key_constructs is not None:
            a["key_constructs"] = list(key_constructs)
        self._coll("alternatives").append(a)
        self._provenance("tf_add_alternative", id)
        return self

    def add_assumption(self, id, statement, added_for=None, protects=None):
        a = {"id": id, "statement": statement, "added_for": added_for}
        if protects is not None:
            a["protects"] = list(protects)
        self._coll("auxiliary_assumptions").append(a)
        self._provenance("tf_add_assumption", id)
        return self

    def set_formal_model(self, type, spec_ref=None):
        self.data["formal_model"] = {"type": type, "spec_ref": spec_ref}
        self._provenance("tf_set_formal_model", type)
        return self

    # -- mirrored public API --------------------------------------------------
    def check(self) -> dict:
        return _check(self.data)

    def report(self, format: str = "json") -> str:
        return _report(self.data, format=format)

    def redundancy_check(self) -> list[dict]:
        return _redundancy_check(self.data)

    def diagram(self, type: str = "nomological_net", engine: str = "graphviz") -> str:
        """Return the diagram IR for ``type`` (one of nomological_net, provenance,
        causal_dag, development_roadmap, pipeline, context, workflow, venn, rigour,
        severity)."""
        return _diagram(self.data, type=type, engine=engine)

    def render_diagram(self, type: str = "nomological_net"):
        """Render a digraph view via the optional ``graphviz`` library; see
        :func:`theoryforge.render.render_diagram`."""
        from .render import render_diagram as _render_diagram
        return _render_diagram(self.data, type=type)

    def severity(self) -> list[dict]:
        """Per-prediction risk and computed severity from the operationalised rubric."""
        return _severity(self.data)

    def appraise_amendment(self, prior) -> dict:
        """Progressive vs degenerating verdict for this theory relative to a prior version."""
        return _appraise_amendment(self.data, prior)

    def implications(self) -> dict:
        """The conditional independencies the causal subgraph commits the theory to."""
        return _implications(self.data)

    def preregister(self, path=None) -> str:
        """Render a preregistration document (and write it if a path is given)."""
        return _preregister(self.data, path)

    def landscape(self, corpus, min_link: int = 2) -> dict:
        """Map this theory and its alternatives onto a literature corpus's themes."""
        return _landscape(self.data, corpus, min_link=min_link)

    def new_evidence_dois(self, candidate_dois: list) -> list:
        """DOIs in `candidate_dois` not already cited by this theory's evidence or alternatives."""
        return _new_evidence_dois(self.data, candidate_dois)

    def compile_sem(self) -> str:
        """Compile constructs and propositions to lavaan model syntax."""
        return _compile_sem(self.data)

    def dossier(self) -> str:
        """A reviewer-facing audit bundle of rigour report, severity, provenance, and preregistration."""
        return _dossier(self.data)

    def simulate(self, steps: int = 10, dt: float = 0.1, k: float = 1.0,
                 damping: float = 0.5, init: float = 1.0) -> dict:
        """Integrate the construct network as a linear dynamical system."""
        return _simulate(self.data, steps=steps, dt=dt, k=k, damping=damping, init=init)

    def embedding_redundancy(self, embedder, threshold=None) -> list[dict]:
        """An opt-in embedding-based redundancy screen; results depend on the supplied embedder."""
        return _embedding_redundancy(self.data, embedder, threshold=threshold)

    def render_report(self, path, title=None, render: bool = False, to: str = "html") -> str:
        """Write (and optionally render) a Quarto report of the audit dossier."""
        return _render_report(self.data, path, title=title, render=render, to=to)

    def osf_push(self, token=None, node=None, filename=None, dry_run: bool = True,
                 base_url: str | None = None) -> dict:
        """Deposit the dossier on OSF (dry-run by default). A live push needs a token and node."""
        kw = {} if base_url is None else {"base_url": base_url}
        return _osf_push(self.data, token=token, node=node, filename=filename, dry_run=dry_run, **kw)

    def __repr__(self) -> str:
        return f"Theory(id={self.id!r}, maturity={self.maturity!r})"

appraise_amendment(prior)

Progressive vs degenerating verdict for this theory relative to a prior version.

Source code in src/theoryforge/core.py

def appraise_amendment(self, prior) -> dict:
    """Progressive vs degenerating verdict for this theory relative to a prior version."""
    return _appraise_amendment(self.data, prior)

compile_sem()

Compile constructs and propositions to lavaan model syntax.

Source code in src/theoryforge/core.py

def compile_sem(self) -> str:
    """Compile constructs and propositions to lavaan model syntax."""
    return _compile_sem(self.data)

diagram(type='nomological_net', engine='graphviz')

Return the diagram IR for type (one of nomological_net, provenance, causal_dag, development_roadmap, pipeline, context, workflow, venn, rigour, severity).

Source code in src/theoryforge/core.py

def diagram(self, type: str = "nomological_net", engine: str = "graphviz") -> str:
    """Return the diagram IR for ``type`` (one of nomological_net, provenance,
    causal_dag, development_roadmap, pipeline, context, workflow, venn, rigour,
    severity)."""
    return _diagram(self.data, type=type, engine=engine)

dossier()

A reviewer-facing audit bundle of rigour report, severity, provenance, and preregistration.

Source code in src/theoryforge/core.py

def dossier(self) -> str:
    """A reviewer-facing audit bundle of rigour report, severity, provenance, and preregistration."""
    return _dossier(self.data)

embedding_redundancy(embedder, threshold=None)

An opt-in embedding-based redundancy screen; results depend on the supplied embedder.

Source code in src/theoryforge/core.py

def embedding_redundancy(self, embedder, threshold=None) -> list[dict]:
    """An opt-in embedding-based redundancy screen; results depend on the supplied embedder."""
    return _embedding_redundancy(self.data, embedder, threshold=threshold)

implications()

The conditional independencies the causal subgraph commits the theory to.

Source code in src/theoryforge/core.py

def implications(self) -> dict:
    """The conditional independencies the causal subgraph commits the theory to."""
    return _implications(self.data)

landscape(corpus, min_link=2)

Map this theory and its alternatives onto a literature corpus's themes.

Source code in src/theoryforge/core.py

def landscape(self, corpus, min_link: int = 2) -> dict:
    """Map this theory and its alternatives onto a literature corpus's themes."""
    return _landscape(self.data, corpus, min_link=min_link)

new_evidence_dois(candidate_dois)

DOIs in candidate_dois not already cited by this theory's evidence or alternatives.

Source code in src/theoryforge/core.py

def new_evidence_dois(self, candidate_dois: list) -> list:
    """DOIs in `candidate_dois` not already cited by this theory's evidence or alternatives."""
    return _new_evidence_dois(self.data, candidate_dois)

osf_push(token=None, node=None, filename=None, dry_run=True, base_url=None)

Deposit the dossier on OSF (dry-run by default). A live push needs a token and node.

Source code in src/theoryforge/core.py

def osf_push(self, token=None, node=None, filename=None, dry_run: bool = True,
             base_url: str | None = None) -> dict:
    """Deposit the dossier on OSF (dry-run by default). A live push needs a token and node."""
    kw = {} if base_url is None else {"base_url": base_url}
    return _osf_push(self.data, token=token, node=node, filename=filename, dry_run=dry_run, **kw)

preregister(path=None)

Render a preregistration document (and write it if a path is given).

Source code in src/theoryforge/core.py

def preregister(self, path=None) -> str:
    """Render a preregistration document (and write it if a path is given)."""
    return _preregister(self.data, path)

render_diagram(type='nomological_net')

Render a digraph view via the optional graphviz library; see :func:theoryforge.render.render_diagram.

Source code in src/theoryforge/core.py

def render_diagram(self, type: str = "nomological_net"):
    """Render a digraph view via the optional ``graphviz`` library; see
    :func:`theoryforge.render.render_diagram`."""
    from .render import render_diagram as _render_diagram
    return _render_diagram(self.data, type=type)

render_report(path, title=None, render=False, to='html')

Write (and optionally render) a Quarto report of the audit dossier.

Source code in src/theoryforge/core.py

def render_report(self, path, title=None, render: bool = False, to: str = "html") -> str:
    """Write (and optionally render) a Quarto report of the audit dossier."""
    return _render_report(self.data, path, title=title, render=render, to=to)

severity()

Per-prediction risk and computed severity from the operationalised rubric.

Source code in src/theoryforge/core.py

def severity(self) -> list[dict]:
    """Per-prediction risk and computed severity from the operationalised rubric."""
    return _severity(self.data)

simulate(steps=10, dt=0.1, k=1.0, damping=0.5, init=1.0)

Integrate the construct network as a linear dynamical system.

Source code in src/theoryforge/core.py

def simulate(self, steps: int = 10, dt: float = 0.1, k: float = 1.0,
             damping: float = 0.5, init: float = 1.0) -> dict:
    """Integrate the construct network as a linear dynamical system."""
    return _simulate(self.data, steps=steps, dt=dt, k=k, damping=damping, init=init)

validate(*, full=False)

Structural validation against the schema's required fields and enums.

Returns True on success, raises ValueError listing every problem found. With full=True additionally checks referential integrity: that every id is unique within its collection and that every cross-reference (proposition endpoints, prediction derivations and diagnostics, assumption/evidence/test-outcome targets) points to a declared id, and that every prediction severity is a number within [0, 1]. The full checks are deterministic.

Source code in src/theoryforge/core.py

def validate(self, *, full: bool = False) -> bool:
    """Structural validation against the schema's required fields and enums.

    Returns True on success, raises ValueError listing every problem found.
    With ``full=True`` additionally checks referential integrity: that every
    id is unique within its collection and that every cross-reference
    (proposition endpoints, prediction derivations and diagnostics,
    assumption/evidence/test-outcome targets) points to a declared id, and
    that every prediction ``severity`` is a number within [0, 1]. The
    ``full`` checks are deterministic.
    """
    errors: list[str] = []
    d = self.data
    for req in ("schema_version", "id", "title", "maturity"):
        if not _nonempty_str(d.get(req)):
            errors.append(f"missing/empty required field: {req}")
    # Each enum test asks whether the value is a nonempty string before
    # asking whether it is a member. A YAML list or mapping reaches these
    # lines whenever a field is mistyped, and `x in <set>` raises TypeError
    # on an unhashable value, which would abandon the collected errors and
    # report something the contract never promised. The string test also
    # keeps the twins level: R's `%in%` coerces a one-element list to its
    # element, so `theory_form: [network]` used to validate there and be
    # refused here.
    mat = d.get("maturity")
    if not (_nonempty_str(mat) and mat in _MATURITY):
        errors.append(f"maturity must be one of {', '.join(sorted(_MATURITY))}")
    if "theory_form" in d:
        form = d["theory_form"]
        if not (_nonempty_str(form) and form in _FORM):
            errors.append(f"theory_form must be one of {', '.join(sorted(_FORM))}")
    # A misspelt collection key would otherwise pass silently and quietly
    # change every downstream verdict (a renamed `predictions:` drops the
    # whole collection), so unrecognised top-level fields are refused.
    known = _resources.theory_schema().get("properties", {})
    for key in d:
        if key not in known:
            errors.append(f"unknown top-level field: {key}")
    for i, c in enumerate(self._list("constructs")):
        for req in ("id", "label", "definition"):
            if not _nonempty_str(_field(c, req)):
                errors.append(f"construct[{i}] missing/empty {req}")
    for i, p in enumerate(self._list("propositions")):
        for req in ("id", "from", "to", "relation"):
            if not _nonempty_str(_field(p, req)):
                errors.append(f"proposition[{i}] missing/empty {req}")
        rel = _field(p, "relation")
        if _nonempty_str(rel) and rel not in _RELATION:
            errors.append(f"proposition[{i}] relation '{rel}' not allowed")
    for i, p in enumerate(self._list("predictions")):
        for req in ("id", "statement", "type"):
            if not _nonempty_str(_field(p, req)):
                errors.append(f"prediction[{i}] missing/empty {req}")
        ty = _field(p, "type")
        if _nonempty_str(ty) and ty not in _PRED_TYPE:
            errors.append(f"prediction[{i}] type '{ty}' not allowed")

    if full:
        self._referential_errors(errors)

    if errors:
        raise ValueError("invalid theory object: " + "; ".join(errors))
    return True

theoryforge.new_theory(id, title, maturity='building', theory_form='network')

Start a new, empty theory object (BUILDING mode entry point).

Source code in src/theoryforge/core.py

def new_theory(id: str, title: str, maturity: str = "building", theory_form: str = "network") -> Theory:
    """Start a new, empty theory object (BUILDING mode entry point)."""
    t = Theory({
        "schema_version": "1.0",
        "id": id,
        "title": title,
        "maturity": maturity,
        "theory_form": theory_form,
    })
    t._provenance("tf_theory", id)
    return t

theoryforge.read(path)

Read a theory object from a YAML or JSON file.

Source code in src/theoryforge/core.py

def read(path) -> Theory:
    """Read a theory object from a YAML or JSON file."""
    path = Path(path)
    text = path.read_text(encoding="utf-8")
    data = json.loads(text) if path.suffix.lower() == ".json" else yaml.safe_load(text)
    if not isinstance(data, dict):
        raise ValueError("Theory data must be a mapping")
    return Theory(data)

theoryforge.write(theory, path)

Write a theory object to YAML or JSON (chosen by file extension).

Source code in src/theoryforge/core.py

def write(theory: Theory, path) -> None:
    """Write a theory object to YAML or JSON (chosen by file extension)."""
    theory.write(path)

Packaged examples

Example theories shipped inside the package, so that nothing has to be downloaded to follow the documentation. The R twin reaches the same files through tf_example_names() and tf_example_path().

theoryforge.example_names()

The names of the example theory and corpus files, sorted.

Only .yaml entries count, matching the R twin's list.files(pattern = "\\.yaml$"). Listing the directory wholesale would let anything else that lands beside the examples, such as a __pycache__ directory left by a local build, show up here and not in R.

Source code in src/theoryforge/examples.py

def example_names() -> list[str]:
    """The names of the example theory and corpus files, sorted.

    Only ``.yaml`` entries count, matching the R twin's ``list.files(pattern =
    "\\\\.yaml$")``. Listing the directory wholesale would let anything else
    that lands beside the examples, such as a ``__pycache__`` directory left by
    a local build, show up here and not in R.
    """
    entries = (files("theoryforge") / "fixtures").iterdir()
    return sorted(p.name for p in entries if p.name.endswith(".yaml"))

theoryforge.example_path(name)

Filesystem path of a packaged example, e.g. panic-network.theory.yaml.

Raises FileNotFoundError naming the available files when name is not one of them.

Source code in src/theoryforge/examples.py

def example_path(name: str) -> Path:
    """Filesystem path of a packaged example, e.g. ``panic-network.theory.yaml``.

    Raises ``FileNotFoundError`` naming the available files when ``name`` is not
    one of them.
    """
    resource = files("theoryforge") / "fixtures" / name
    if not resource.is_file():
        raise FileNotFoundError(f"no packaged example named {name!r}; available: {example_names()}")
    # A wheel is a directory on disk once installed, so this resolves to a real
    # path; as_file() keeps the call correct for a zipped install too.
    with as_file(resource) as path:
        return Path(path)

Rigour

Score a theory against the versioned rigour checklist, and report what the score was made of.

theoryforge.check(T)

Compute the full rigour report (dict) for a Theory or theory mapping.

Source code in src/theoryforge/rigor.py

def check(T) -> dict:
    """Compute the full rigour report (dict) for a Theory or theory mapping."""
    T = T.data if hasattr(T, "data") else T
    spec = _resources.checklist()
    thr = spec["thresholds"]
    results = _check_items(T, thr)

    items = []
    weighted = 0.0
    n_blockers_failed = 0
    for spec_item in spec["items"]:
        iid = spec_item["id"]
        status, score = results[iid]
        weighted += spec_item["weight"] * score
        if spec_item["severity_if_fail"] == "blocker" and status == "fail":
            n_blockers_failed += 1
        items.append({
            "id": iid,
            "status": status,
            "score": score,
            "weight": spec_item["weight"],
            "severity_if_fail": spec_item["severity_if_fail"],
            "citation": spec_item["citation"],
        })

    # Explicit nulls read as "" (the R twin's .tf_str reading). dict.get's
    # default alone covers only absent keys, and the report is compared
    # semantically across the twins, so a null here must not reach the output.
    maturity = T.get("maturity") or ""
    if maturity == "draft":
        gate = "advisory"
    else:
        gate = "blocked" if n_blockers_failed > 0 else "pass"

    return {
        "theory_id": T.get("id") or "",
        "schema_version": T.get("schema_version") or "",
        # Every number below comes from the checklist's weights and thresholds,
        # so two reports are only comparable if they were scored against the
        # same checklist. `schema_version` above is the theory's, not this.
        "checklist_version": spec.get("schema_version") or "",
        "maturity": maturity,
        "aggregate_score": rnd(weighted * 100, 1),
        "gate": gate,
        "n_blockers_failed": n_blockers_failed,
        "items": items,
    }

theoryforge.report(T, format='json')

Render the rigour report. format in {'json', 'html'}.

Source code in src/theoryforge/rigor.py

def report(T, format: str = "json") -> str:
    """Render the rigour report. format in {'json', 'html'}."""
    rep = check(T)  # check() unwraps Theory -> mapping
    if format == "json":
        return json.dumps(rep, indent=2, ensure_ascii=False)
    if format == "html":
        rows = "\n".join(
            f'    <tr><td>{it["id"]}</td><td>{it["status"]}</td>'
            f'<td>{it["score"]}</td><td>{it["citation"]}</td></tr>'
            for it in rep["items"]
        )
        return (
            f'<section class="theoryforge-report">\n'
            f'  <h2>Rigour report: {rep["theory_id"]}</h2>\n'
            f'  <p>Aggregate score: <strong>{rep["aggregate_score"]}</strong> &middot; '
            f'gate: <strong>{rep["gate"]}</strong></p>\n'
            f'  <table>\n    <tr><th>item</th><th>status</th><th>score</th><th>grounding</th></tr>\n'
            f'{rows}\n  </table>\n</section>\n'
        )
    raise ValueError(f"unknown report format: {format!r}")

theoryforge.severity(T)

Per-prediction risk and computed severity, in file order.

References

Mayo, D. G. (2018). Statistical inference as severe testing. Cambridge University Press. https://doi.org/10.1017/9781107286184 Meehl, P. E. (1990). Why summaries of research on psychological theories are often uninterpretable. Psychological Reports, 66, 195-244. https://doi.org/10.2466/pr0.1990.66.1.195

Source code in src/theoryforge/scoring.py

def severity(T) -> list[dict]:
    """Per-prediction risk and computed severity, in file order.

    References:
        Mayo, D. G. (2018). Statistical inference as severe testing. Cambridge
        University Press. https://doi.org/10.1017/9781107286184
        Meehl, P. E. (1990). Why summaries of research on psychological theories
        are often uninterpretable. Psychological Reports, 66, 195-244.
        https://doi.org/10.2466/pr0.1990.66.1.195
    """
    T = T.data if hasattr(T, "data") else T
    preds = _list(T, "predictions")
    alt_ids = {a.get("id") for a in _list(T, "alternatives")}
    out = []
    for p in preds:
        typ = p.get("type")
        base = BASE.get(typ, 0.0)
        discounted = base * (1 - CRUD) if typ == "directional" else base
        dv = _as_list(p.get("diagnostic_vs"))
        diag_bonus = 0.1 if dv and any(d in alt_ids for d in dv) else 0.0
        out.append({
            "prediction_id": p.get("id"),
            "type": typ,
            "risk_score": rnd(base, 3),
            "computed_severity": rnd(min(1.0, discounted + diag_bonus), 3),
        })
    return out

Redundancy

Screen a theory's constructs for lexical redundancy against each other and against the field, deterministically, with an opt-in embedding variant.

theoryforge.tokens(s)

Tokenise a string into a set of content tokens.

Source code in src/theoryforge/redundancy.py

def tokens(s: str) -> set[str]:
    """Tokenise a string into a set of content tokens."""
    s = (s or "").lower()
    parts = _NON_ALNUM.sub(" ", s).split()
    return {t for t in parts if len(t) >= 3 and t not in STOPWORDS}

theoryforge.jaccard(a, b)

Source code in src/theoryforge/redundancy.py

def jaccard(a: set[str], b: set[str]) -> float:
    if not a and not b:
        return 0.0
    inter = len(a & b)
    union = len(a | b)
    return rnd(inter / union, 3)

theoryforge.redundancy_check(T)

Pairwise lexical similarity of construct definitions.

Returns one record per unordered construct pair, sorted by descending similarity then (a, b) ascending.

References

Le, H., Schmidt, F. L., Harter, J. K., & Lauver, K. J. (2010). The problem of empirical redundancy of constructs. Organizational Behavior and Human Decision Processes, 112(2), 112-125. https://doi.org/10.1016/j.obhdp.2010.02.003 Lawson, K. M., & Robins, R. W. (2021). Sibling constructs. Personality and Social Psychology Review, 25(4), 344-366. https://doi.org/10.1177/10888683211047101

Source code in src/theoryforge/redundancy.py

def redundancy_check(T: dict) -> list[dict]:
    """Pairwise lexical similarity of construct definitions.

    Returns one record per unordered construct pair, sorted by descending
    similarity then ``(a, b)`` ascending.

    References:
        Le, H., Schmidt, F. L., Harter, J. K., & Lauver, K. J. (2010). The
        problem of empirical redundancy of constructs. Organizational Behavior
        and Human Decision Processes, 112(2), 112-125.
        https://doi.org/10.1016/j.obhdp.2010.02.003
        Lawson, K. M., & Robins, R. W. (2021). Sibling constructs. Personality
        and Social Psychology Review, 25(4), 344-366.
        https://doi.org/10.1177/10888683211047101
    """
    T = T.data if hasattr(T, "data") else T
    cons = _list(T, "constructs")
    thr = _resources.checklist()["thresholds"]["redundancy_similarity_max"]
    toks = [(c.get("id", ""), tokens(c.get("definition", ""))) for c in cons]
    rows = []
    for i in range(len(toks)):
        for j in range(i + 1, len(toks)):
            sim = jaccard(toks[i][1], toks[j][1])
            rows.append({
                "a": toks[i][0],
                "b": toks[j][0],
                "similarity": sim,
                "flag": "review" if sim >= thr else "ok",
            })
    rows.sort(key=lambda r: (-r["similarity"], r["a"], r["b"]))
    return rows

theoryforge.embedding_redundancy(T, embedder, threshold=None)

Pairwise cosine similarity of embedded construct definitions.

embedder maps a definition string to a numeric vector; the vectors of every compared pair must be of equal, nonzero length, or the pair is refused. Returns one record per unordered construct pair, sorted by descending similarity then (a, b), with a review/ok flag.

Source code in src/theoryforge/embedding.py

def embedding_redundancy(T, embedder: Callable[[str], Sequence[float]],
                         threshold: float | None = None) -> list[dict]:
    """Pairwise cosine similarity of embedded construct definitions.

    `embedder` maps a definition string to a numeric vector; the vectors of every compared
    pair must be of equal, nonzero length, or the pair is refused. Returns one record per
    unordered construct pair, sorted by descending similarity then (a, b), with a
    `review`/`ok` flag.
    """
    T = T.data if hasattr(T, "data") else T
    if threshold is None:
        threshold = _resources.checklist()["thresholds"]["redundancy_similarity_max"]
    cons = _list(T, "constructs")
    vecs = [(c.get("id", ""), embedder(c.get("definition", ""))) for c in cons]
    rows = []
    for i in range(len(vecs)):
        for j in range(i + 1, len(vecs)):
            # Unequal or empty vectors have no defensible cosine, and the two
            # engines read them differently by accident (R recycled the
            # shorter vector, Python truncated the longer), so the same
            # embedder produced two confident, different similarities. Refuse
            # instead of picking a reading.
            la, lb = len(vecs[i][1]), len(vecs[j][1])
            if la == 0 or lb == 0 or la != lb:
                raise ValueError(
                    "embedding_redundancy requires equal-length nonempty embedding vectors; "
                    f"constructs {vecs[i][0] or ''} and {vecs[j][0] or ''} "
                    f"have lengths {la} and {lb}"
                )
            sim = rnd(_cosine(vecs[i][1], vecs[j][1]), 6)
            rows.append({
                "a": vecs[i][0], "b": vecs[j][0],
                "cosine": sim, "flag": "review" if sim >= threshold else "ok",
            })
    rows.sort(key=lambda r: (-r["cosine"], r["a"], r["b"]))
    return rows

Diagram

Byte-identical diagram intermediate representations, and native rendering of them.

theoryforge.diagram.diagram(T, type='nomological_net', engine='graphviz')

Return the diagram IR string for the requested type.

engine is accepted but has no effect, because the IR is engine-independent (DOT for the digraphs, dagitty syntax for the causal DAG, and SVG for the Venn, the rigour grid and the severity chart).

Source code in src/theoryforge/diagram.py

def diagram(T: dict, type: str = "nomological_net", engine: str = "graphviz") -> str:
    """Return the diagram IR string for the requested type.

    ``engine`` is accepted but has no effect, because the IR is engine-independent
    (DOT for the digraphs, dagitty syntax for the causal DAG, and SVG for the Venn,
    the rigour grid and the severity chart).
    """
    T = T.data if hasattr(T, "data") else T
    if type == "nomological_net":
        return _nomological_net(T)
    if type == "provenance":
        return _provenance(T)
    if type == "causal_dag":
        return _causal_dag(T)
    if type == "development_roadmap":
        return _development_roadmap(T)
    if type == "pipeline":
        return _pipeline(T)
    if type == "context":
        return _context(T)
    if type == "workflow":
        return _workflow(T)
    if type == "venn":
        return _venn(T)
    if type == "rigour":
        return _rigor(T)
    if type == "severity":
        return _severity_chart(T)
    raise ValueError(f"unknown diagram type {type!r}; expected one of {_TYPES}")

theoryforge.lit_diagram(obj, type='keyword_cooccurrence')

DOT for the literature layer. type in {keyword_cooccurrence, co_citation, theme_landscape}.

Source code in src/theoryforge/lit.py

def lit_diagram(obj: dict, type: str = "keyword_cooccurrence") -> str:
    """DOT for the literature layer. type in {keyword_cooccurrence, co_citation, theme_landscape}."""
    if type in ("keyword_cooccurrence", "co_citation"):
        return _undirected(type, obj.get(type, []))
    if type == "theme_landscape":
        return _theme_landscape(obj)
    raise ValueError(
        f"unknown lit diagram type {type!r}; expected one of "
        "('keyword_cooccurrence', 'co_citation', 'theme_landscape')"
    )

theoryforge.render_diagram(x, type='nomological_net')

Render a digraph view without leaving Python.

Where :func:theoryforge.diagram.diagram returns the deterministic Graphviz DOT string, render_diagram wraps it in a graphviz.Source, which displays inline in Jupyter and renders to a file with its render method. Requires the optional graphviz <https://pypi.org/project/graphviz/>_ library (pip install theoryforge[render]) and, to write image files, the Graphviz system binaries.

The three chart views (venn, rigour and severity) are already SVG, so they come back as an :class:SVGString, which also displays inline in Jupyter. The causal_dag view emits dagitty syntax rather than DOT, so it is not rendered here; paste it into a dagitty tool instead.

Parameters

x: A :class:~theoryforge.core.Theory, a parsed theory mapping, or a diagram IR string from diagram() or lit_diagram(), so literature diagrams render the same way. type: The diagram type, as in diagram(). Ignored when x is already an IR string.

Returns

graphviz.Source or SVGString A graphviz.Source for the digraph views; an :class:SVGString for the chart views.

Source code in src/theoryforge/render.py

def render_diagram(x, type: str = "nomological_net"):
    """Render a digraph view without leaving Python.

    Where :func:`theoryforge.diagram.diagram` returns the deterministic Graphviz
    DOT string, ``render_diagram`` wraps it in a ``graphviz.Source``, which
    displays inline in Jupyter and renders to a file with its ``render`` method.
    Requires the optional `graphviz <https://pypi.org/project/graphviz/>`_
    library (``pip install theoryforge[render]``) and, to write image files, the
    Graphviz system binaries.

    The three chart views (``venn``, ``rigour`` and ``severity``) are already
    SVG, so they come back as an :class:`SVGString`, which also displays inline
    in Jupyter. The ``causal_dag`` view emits dagitty syntax rather than DOT, so
    it is not rendered here; paste it into a dagitty tool instead.

    Parameters
    ----------
    x:
        A :class:`~theoryforge.core.Theory`, a parsed theory mapping, or a
        diagram IR string from ``diagram()`` or ``lit_diagram()``, so literature
        diagrams render the same way.
    type:
        The diagram type, as in ``diagram()``. Ignored when ``x`` is already an
        IR string.

    Returns
    -------
    graphviz.Source or SVGString
        A ``graphviz.Source`` for the digraph views; an :class:`SVGString` for
        the chart views.
    """
    if isinstance(x, str):
        ir = x
        if ir.lstrip().startswith("<svg"):
            return SVGString(ir)
        if ir.lstrip().startswith("dag"):
            raise ValueError(
                "this is dagitty syntax (the causal_dag view), not Graphviz DOT; "
                "render it with a dagitty tool such as dagitty.net."
            )
    else:
        if type == "causal_dag":
            raise ValueError(
                "the causal_dag view emits dagitty syntax, not Graphviz DOT; "
                "render diagram(x, 'causal_dag') with a dagitty tool such as "
                "dagitty.net."
            )
        if type in _SVG_TYPES:
            return SVGString(_diagram(x, type=type))
        if type not in _DOT_TYPES:
            # Delegate to diagram() for its canonical unknown-type error.
            _diagram(x, type=type)
        ir = _diagram(x, type=type)

    try:
        import graphviz
    except ImportError as err:
        raise ImportError(
            "render_diagram() needs the optional 'graphviz' library; install it "
            "with pip install theoryforge[render], or use diagram() and render "
            "the DOT string with any Graphviz tool."
        ) from err
    return graphviz.Source(ir)

Develop

Appraise an amendment to a theory as progressive or degenerating, in Lakatos's sense.

theoryforge.appraise_amendment(new, prior)

Appraise an amendment as progressive, degenerating, or neutral relative to a prior version.

References

Lakatos, I. (1970). Falsification and the methodology of scientific research programmes. In Criticism and the growth of knowledge (pp. 91-196). Cambridge University Press. https://doi.org/10.1017/cbo9781139171434.009 Meehl, P. E. (1990). Appraising and amending theories. Psychological Inquiry, 1(2), 108-141. https://doi.org/10.1207/s15327965pli0102_1

Source code in src/theoryforge/develop.py

def appraise_amendment(new, prior) -> dict:
    """Appraise an amendment as progressive, degenerating, or neutral relative to a prior version.

    References:
        Lakatos, I. (1970). Falsification and the methodology of scientific
        research programmes. In Criticism and the growth of knowledge
        (pp. 91-196). Cambridge University Press.
        https://doi.org/10.1017/cbo9781139171434.009
        Meehl, P. E. (1990). Appraising and amending theories. Psychological
        Inquiry, 1(2), 108-141. https://doi.org/10.1207/s15327965pli0102_1
    """
    new = new.data if hasattr(new, "data") else new
    prior = prior.data if hasattr(prior, "data") else prior

    prior_pred_ids = {p.get("id") for p in _list(prior, "predictions")}
    prior_aux_ids = {a.get("id") for a in _list(prior, "auxiliary_assumptions")}
    tos = _list(new, "test_outcomes")

    def passed(pid: str) -> bool:
        return any(t.get("prediction_id") == pid and t.get("passed") is True for t in tos)

    new_predictions = [p.get("id") for p in _list(new, "predictions") if p.get("id") not in prior_pred_ids]
    corroborated_new = [pid for pid in new_predictions if passed(pid)]

    ad_hoc = []
    for a in _list(new, "auxiliary_assumptions"):
        if a.get("id") in prior_aux_ids:
            continue
        if a.get("added_for") is None:
            continue
        protects = _as_list(a.get("protects"))
        if not any(t.get("prediction_id") in protects and t.get("passed") is True for t in tos):
            ad_hoc.append(a.get("id"))

    if len(corroborated_new) >= 1 and len(ad_hoc) == 0:
        verdict = "progressive"
    elif len(ad_hoc) >= 1 and len(corroborated_new) == 0:
        verdict = "degenerating"
    else:
        verdict = "neutral"

    return {
        "verdict": verdict,
        "new_predictions": sorted(new_predictions),
        "corroborated_new": sorted(corroborated_new),
        "ad_hoc_assumptions": sorted(ad_hoc),
    }

Testing and review

The conditional independencies a causal theory commits to, preregistration, SEM compilation and the reviewer-facing audit dossier.

theoryforge.implications.implications(T)

Conditional independencies implied by a theory's causal subgraph.

Reads the propositions whose relation is causal (causes, increases, decreases) as directed edges over the constructs they connect, checks that the resulting graph is acyclic, and returns its basis set: for every pair of non-adjacent constructs, the claim that the two are independent given the parents of both (Pearl, 1988; Shipley, 2000). A basis set implies every other conditional independence the graph entails, so it is the shortest complete statement of what the theory forbids in data.

Returns {theory_id, acyclic, constructs, n_edges, implications, n_implications}. constructs lists, in file order, the constructs that a causal proposition connects; constructs the theory says nothing causal about are left out, because silence about a construct is not a claim that it is independent of anything. acyclic is always True in a returned record, since a cyclic graph is refused, and is carried so that a serialised record states the verdict rather than leaving a reader to infer that the check ran. Each entry of implications is {a, b, given, statement}, where statement renders the claim in the notation dagitty prints, a _||_ b | z1, z2. Pairs come in construct file order, as do the members of given, so the two engines return the same records in the same order.

A theory with no causal propositions comes back with an empty basis set and no error: constructs and implications are empty and n_implications is 0.

Raises:

Type	Description
ValueError	if two constructs share an id, if a causal proposition names a construct the theory has not declared, or if the causal graph has a cycle, in which case no basis set is defined and the message names a cycle that was found.

References

Pearl, J. (1988). Probabilistic reasoning in intelligent systems: Networks of plausible inference. Morgan Kaufmann. Shipley, B. (2000). A new inferential test for path models based on directed acyclic graphs. Structural Equation Modeling, 7(2), 206-218. https://doi.org/10.1207/S15328007SEM0702_4

Example

A mediated chain, arousal raising perceived threat and perceived threat raising avoidance, commits the theory to one thing it does not state directly, that arousal and avoidance are independent once perceived threat is held fixed.

import theoryforge as tf

t = tf.new_theory("mediation", "A mediated chain")
t.add_construct("c_arousal", "Arousal", "Bodily activation.")
t.add_construct("c_threat", "Perceived threat", "Appraised danger.")
t.add_construct("c_avoidance", "Avoidance", "Withdrawal from the trigger.")
t.add_proposition("p1", "c_arousal", "c_threat", "increases")
t.add_proposition("p2", "c_threat", "c_avoidance", "increases")

[i["statement"] for i in t.implications()["implications"]]
# ['c_arousal _||_ c_avoidance | c_threat']

Source code in src/theoryforge/implications.py

def implications(T) -> dict:
    """Conditional independencies implied by a theory's causal subgraph.

    Reads the propositions whose relation is causal (``causes``, ``increases``,
    ``decreases``) as directed edges over the constructs they connect, checks
    that the resulting graph is acyclic, and returns its basis set: for every
    pair of non-adjacent constructs, the claim that the two are independent
    given the parents of both (Pearl, 1988; Shipley, 2000). A basis set implies
    every other conditional independence the graph entails, so it is the shortest
    complete statement of what the theory forbids in data.

    Returns ``{theory_id, acyclic, constructs, n_edges, implications,
    n_implications}``. ``constructs`` lists, in file order, the constructs that
    a causal proposition connects; constructs the theory says nothing causal
    about are left out, because silence about a construct is not a claim that it
    is independent of anything. ``acyclic`` is always True in a returned record,
    since a cyclic graph is refused, and is carried so that a serialised record
    states the verdict rather than leaving a reader to infer that the check ran.
    Each entry of ``implications`` is ``{a, b, given, statement}``, where
    ``statement`` renders the claim in the notation dagitty prints,
    ``a _||_ b | z1, z2``. Pairs come in construct file order, as do the members
    of ``given``, so the two engines return the same records in the same order.

    A theory with no causal propositions comes back with an empty basis set and
    no error: ``constructs`` and ``implications`` are empty and ``n_implications``
    is 0.

    Raises:
        ValueError: if two constructs share an id, if a causal proposition names
            a construct the theory has not declared, or if the causal graph has a
            cycle, in which case no basis set is defined and the message names a
            cycle that was found.

    References:
        Pearl, J. (1988). Probabilistic reasoning in intelligent systems:
        Networks of plausible inference. Morgan Kaufmann.
        Shipley, B. (2000). A new inferential test for path models based on
        directed acyclic graphs. Structural Equation Modeling, 7(2), 206-218.
        https://doi.org/10.1207/S15328007SEM0702_4

    Example:
        A mediated chain, arousal raising perceived threat and perceived threat
        raising avoidance, commits the theory to one thing it does not state
        directly, that arousal and avoidance are independent once perceived
        threat is held fixed.

        ```python
        import theoryforge as tf

        t = tf.new_theory("mediation", "A mediated chain")
        t.add_construct("c_arousal", "Arousal", "Bodily activation.")
        t.add_construct("c_threat", "Perceived threat", "Appraised danger.")
        t.add_construct("c_avoidance", "Avoidance", "Withdrawal from the trigger.")
        t.add_proposition("p1", "c_arousal", "c_threat", "increases")
        t.add_proposition("p2", "c_threat", "c_avoidance", "increases")

        [i["statement"] for i in t.implications()["implications"]]
        # ['c_arousal _||_ c_avoidance | c_threat']
        ```
    """
    T = T.data if hasattr(T, "data") else T

    declared: list[str] = []
    position: dict[str, int] = {}
    for c in _list(T, "constructs"):
        cid = _str(c.get("id"))
        # Two constructs sharing an id give the same node two sets of parents,
        # and nothing in the maths says which one a proposition meant.
        if cid in position:
            raise ValueError(
                f"implications requires unique construct ids; duplicate construct id: {cid}")
        position[cid] = len(declared)
        declared.append(cid)

    edges: list[tuple[int, int]] = []
    for p in _list(T, "propositions"):
        if p.get("relation") not in _CAUSAL:
            continue
        pid = _str(p.get("id"))
        frm, to = _str(p.get("from")), _str(p.get("to"))
        # Dropping an edge whose endpoint was never declared would shrink the
        # graph and so add independencies the theory does not imply, which is a
        # confidently wrong answer rather than a missing one.
        for endpoint in (frm, to):
            if endpoint not in position:
                raise ValueError(
                    "implications requires causal propositions between declared constructs; "
                    f"proposition '{pid}' refers to unknown construct '{endpoint}'")
        e = (position[frm], position[to])
        if e not in edges:
            edges.append(e)

    used = sorted({i for e in edges for i in e})
    nodes = [declared[i] for i in used]
    k = len(nodes)
    rank = {i: r for r, i in enumerate(used)}
    adj = [[False] * k for _ in range(k)]
    for u, v in edges:
        adj[rank[u]][rank[v]] = True

    cycle = _first_cycle(adj, k)
    if cycle is not None:
        raise ValueError(
            "implications requires an acyclic causal graph; cycle found: "
            + " -> ".join(nodes[i] for i in cycle))

    parents = [[j for j in range(k) if adj[j][i]] for i in range(k)]
    out: list[dict] = []
    for i in range(k):
        for j in range(i + 1, k):
            if adj[i][j] or adj[j][i]:
                continue
            # In a DAG neither member of a non-adjacent pair can be a parent of
            # the other, so the union needs no further exclusion.
            given = [nodes[g] for g in sorted(set(parents[i]) | set(parents[j]))]
            out.append({"a": nodes[i], "b": nodes[j], "given": given,
                        "statement": _statement(nodes[i], nodes[j], given)})

    return {
        "theory_id": _str(T.get("id")),
        "acyclic": True,
        "constructs": nodes,
        "n_edges": len(edges),
        "implications": out,
        "n_implications": len(out),
    }

theoryforge.preregister(T, path=None)

Render a preregistration document, writing it to path when one is given.

Source code in src/theoryforge/prereg.py

def preregister(T, path=None) -> str:
    """Render a preregistration document, writing it to ``path`` when one is given."""
    data = T.data if hasattr(T, "data") else T
    rep = _check(data)
    deriv = next((it for it in rep["items"] if it["id"] == "derivation_chain"), None)
    verified = "yes" if deriv and deriv["status"] == "pass" else "no"

    lines = [
        f"# Preregistration: {data.get('title', '')}",
        "",
        f"- Theory ID: {data.get('id', '')}",
        f"- Schema version: {data.get('schema_version', '')}",
        f"- Maturity: {data.get('maturity', '')}",
        f"- Derivation chain verified: {verified}",
        "",
        "## Hypotheses",
    ]
    preds = _list(data, "predictions")
    if not preds:
        lines.append("_No predictions specified._")
    else:
        for i, p in enumerate(preds, start=1):
            df = _as_list(p.get("derives_from"))
            df_txt = ", ".join(df) if df else "—"
            lines.append(f"{i}. [{p.get('type')}] {p.get('statement')} (derives from: {df_txt})")

    lines += ["", "## Severity"]
    sev = _severity(data)
    if not sev:
        lines.append("_No predictions specified._")
    else:
        for s in sev:
            pid, cs, rk = s["prediction_id"], _fmt(s["computed_severity"]), _fmt(s["risk_score"])
            lines.append(f"- {pid}: severity {cs}, risk {rk}")

    text = "\n".join(lines) + "\n"
    if path is not None:
        _write_lf(path, text)
    return text

theoryforge.compile_sem(T)

Source code in src/theoryforge/sem.py

def compile_sem(T) -> str:
    T = T.data if hasattr(T, "data") else T
    out = [f"# lavaan model generated by theoryforge for {T.get('id', '')}", "# Measurement model"]
    for c in _list(T, "constructs"):
        meas = _as_list(c.get("measurement"))
        if meas:
            out.append(f"{c.get('id')} =~ " + " + ".join(_san(m) for m in meas))
    out.append("# Structural model")
    for p in _list(T, "propositions"):
        rel, frm, to = p.get("relation"), p.get("from"), p.get("to")
        if rel in _PATH:
            out.append(f"{to} ~ {frm}")
        elif rel == "associates":
            out.append(f"{frm} ~~ {to}")
        elif rel == "moderates":
            out.append(f"# moderation: {frm} moderates the path into {to} (specify interaction manually)")
    return "\n".join(out) + "\n"

theoryforge.dossier.dossier(T)

Source code in src/theoryforge/dossier.py

def dossier(T) -> str:
    data = T.data if hasattr(T, "data") else T
    rep = _check(data)
    lines = [
        f"# theoryforge dossier: {data.get('title', '')}",
        "",
        f"- Theory ID: {data.get('id', '')}",
        f"- Maturity: {data.get('maturity', '')}",
        # The score is only interpretable against the checklist that produced
        # it, so a reviewer reading the bundle can see which one that was.
        f"- Checklist version: {rep['checklist_version']}",
        f"- Aggregate rigour score: {_fmt(rep['aggregate_score'])}/100",
        f"- Gate: {rep['gate']}",
        f"- Blockers failed: {rep['n_blockers_failed']}",
        "",
        "## Rigour checklist",
        "",
        "| item | status | score | weight |",
        "| --- | --- | --- | --- |",
    ]
    for it in rep["items"]:
        lines.append(f"| {it['id']} | {it['status']} | {_fmt(it['score'])} | {_fmt(it['weight'])} |")

    lines += ["", "## Severity", ""]
    sev = _severity(data)
    if not sev:
        lines.append("_No predictions specified._")
    else:
        for s in sev:
            pid, cs, rk = s["prediction_id"], _fmt(s["computed_severity"]), _fmt(s["risk_score"])
            lines.append(f"- {pid}: severity {cs}, risk {rk}")

    lines += ["", "## Provenance", ""]
    prov = _list(data, "provenance")
    if not prov:
        lines.append("_No provenance recorded._")
    else:
        for i, s in enumerate(prov, start=1):
            action = str(s.get("action", "") or "")
            detail = str(s.get("detail", "") or "")
            lines.append(f"{i}. {action}: {detail}" if detail.strip() else f"{i}. {action}")

    lines += ["", "## Preregistration", ""]
    return "\n".join(lines) + "\n" + _preregister(data)

Simulation

Integrate the construct network as a deterministic dynamical system.

theoryforge.simulate.simulate(T, steps=10, dt=0.1, k=1.0, damping=0.5, init=1.0)

Integrate the theory's construct network as a linear dynamical system.

Returns {states, dt, steps, k, damping, init, trajectory}, where trajectory[t] is the state vector at step t (t = 0..steps), each value rounded to 6 decimals. All five knobs are echoed back, because the trajectory cannot be reproduced without them.

Source code in src/theoryforge/simulate.py

def simulate(T, steps: int = 10, dt: float = 0.1, k: float = 1.0,
             damping: float = 0.5, init: float = 1.0) -> dict:
    """Integrate the theory's construct network as a linear dynamical system.

    Returns {states, dt, steps, k, damping, init, trajectory}, where trajectory[t] is the
    state vector at step t (t = 0..steps), each value rounded to 6 decimals. All five
    knobs are echoed back, because the trajectory cannot be reproduced without them.
    """
    T = T.data if hasattr(T, "data") else T
    states = [c.get("id") for c in _list(T, "constructs")]
    n = len(states)
    # Duplicate construct ids have no defensible reading here, and the two
    # engines resolved them differently by accident (a dict comprehension keeps
    # the last index, R's `[[` on a named vector the first), so the same file
    # produced two plausible trajectories. Refuse instead of picking a winner.
    seen: set = set()
    for s in states:
        if s in seen:
            raise ValueError(f"simulate requires unique construct ids; duplicate construct id: {s}")
        seen.add(s)
    idx = {s: i for i, s in enumerate(states)}

    A = [[0.0] * n for _ in range(n)]
    for p in _list(T, "propositions"):
        f, t, rel = p.get("from"), p.get("to"), p.get("relation")
        if f in idx and t in idx:
            sign = 1.0 if rel in _POS else (-1.0 if rel in _NEG else 0.0)
            A[idx[t]][idx[f]] += sign * k

    X = [float(init)] * n
    traj = [[rnd(x, 6) for x in X]]
    for _ in range(steps):
        dX = [sum(A[i][j] * X[j] for j in range(n)) - damping * X[i] for i in range(n)]
        X = [X[i] + dt * dX[i] for i in range(n)]
        traj.append([rnd(x, 6) for x in X])

    return {"states": states, "dt": dt, "steps": steps, "k": k, "damping": damping,
            "init": init, "trajectory": traj}

Reporting and deposit

Render a report from a theory, and deposit the result.

theoryforge.render_report(T, path, title=None, render=False, to='html')

Write a Quarto report for the theory; render it with Quarto when render=True.

Returns the path of the written .qmd.

Source code in src/theoryforge/report_render.py

def render_report(T, path, title: str | None = None, render: bool = False, to: str = "html") -> str:
    """Write a Quarto report for the theory; render it with Quarto when ``render=True``.

    Returns the path of the written `.qmd`.
    """
    data = T.data if hasattr(T, "data") else T
    # Fall back to the id when the title is empty as well as absent (matches R's nzchar fallback).
    title = title or f"theoryforge report: {data.get('title') or data.get('id') or ''}"
    title = title.replace('"', "'")
    path = Path(path)
    if path.suffix.lower() != ".qmd":
        path = path.with_suffix(".qmd")
    header = f'---\ntitle: "{title}"\nformat: {to}\n---\n\n'
    _write_lf(path, header + _dossier(data))
    if render:
        subprocess.run(["quarto", "render", str(path), "--to", to], check=True)  # pragma: no cover
    return str(path)

theoryforge.osf_push(T, token=None, node=None, filename=None, dry_run=True, base_url=_DEFAULT_BASE)

Deposit the theory's dossier on OSF.

With dry_run=True (default) the planned request is returned and nothing is sent. A live upload (dry_run=False) requires both token and node (the OSF project id).

Source code in src/theoryforge/osf.py

def osf_push(T, token: str | None = None, node: str | None = None,
             filename: str | None = None, dry_run: bool = True,
             base_url: str = _DEFAULT_BASE) -> dict:
    """Deposit the theory's dossier on OSF.

    With ``dry_run=True`` (default) the planned request is returned and nothing is sent. A live
    upload (``dry_run=False``) requires both ``token`` and ``node`` (the OSF project id).
    """
    data = T.data if hasattr(T, "data") else T
    # A null or empty id must not leak into the filename ('None.dossier.md' /
    # '.dossier.md'); fall back to 'theory' as the R twin's nzchar guard does.
    tid = data.get("id") or "theory"
    fname = filename or f"{tid}.dossier.md"
    content = _dossier(data)
    # Percent-encode the filename (theory ids are user-supplied, so fname may
    # carry spaces, '&' or '#'); mirrors the R utils::URLencode(reserved = TRUE)
    # call so the dry-run request dicts stay parity-identical.
    url = (
        f"{base_url}{node}/providers/osfstorage/?kind=file&name={_quote(fname, safe='')}"
        if node else None
    )
    request = {"method": "PUT", "url": url, "filename": fname,
               "content_bytes": len(content.encode("utf-8"))}

    if dry_run:
        return {"dry_run": True, "request": request,
                "note": "set dry_run=False with a valid token and node to perform the upload"}
    if not token or not node:
        raise ValueError("a live OSF push requires both `token` and `node` (the OSF project id)")
    assert url is not None  # node is set, so url was constructed above
    import urllib.request  # pragma: no cover - network path, not exercised in tests
    req = urllib.request.Request(
        url, data=content.encode("utf-8"), method="PUT",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "text/markdown"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:  # pragma: no cover - network
        return {"dry_run": False, "status": getattr(resp, "status", None), "filename": fname}

Literature layer

Map the literature a theory sits in, and check what of it the theory does not yet cite.

theoryforge.read_corpus(path)

Read a literature corpus from YAML or JSON.

Source code in src/theoryforge/lit.py

def read_corpus(path) -> dict:
    """Read a literature corpus from YAML or JSON."""
    path = Path(path)
    text = path.read_text(encoding="utf-8")
    data = json.loads(text) if path.suffix.lower() == ".json" else yaml.safe_load(text)
    if not isinstance(data, dict):
        raise ValueError("Corpus data must be a mapping")
    return data

theoryforge.litmap(corpus, min_link=DEFAULT_MIN_LINK)

Keyword co-occurrence, thematic components, and co-citation, all deterministic.

Source code in src/theoryforge/lit.py

def litmap(corpus, min_link: int = DEFAULT_MIN_LINK) -> dict:
    """Keyword co-occurrence, thematic components, and co-citation, all deterministic."""
    records = _records(corpus)
    all_keywords = sorted({k for r in records for k in _as_list(r.get("keywords")) if k})
    kw_edges = _edges(_pair_counts(records, "keywords"), min_link)
    cocit = _edges(_pair_counts(records, "references"), min_link)
    return {
        "n_records": len(records),
        "keywords": all_keywords,
        "keyword_cooccurrence": kw_edges,
        "themes": _components(kw_edges),
        "co_citation": cocit,
    }

theoryforge.landscape(theory, corpus, min_link=DEFAULT_MIN_LINK)

Map a theory and its registered alternatives onto the literature's themes.

Source code in src/theoryforge/lit.py

def landscape(theory, corpus, min_link: int = DEFAULT_MIN_LINK) -> dict:
    """Map a theory and its registered alternatives onto the literature's themes."""
    T = theory.data if hasattr(theory, "data") else theory
    lm = litmap(corpus, min_link)

    focal_src = " ".join(
        [T.get("title", "")] + [c.get("label", "") for c in (T.get("constructs") or [])]
    )
    focal_tokens = tokens(focal_src)
    alts = T.get("alternatives") or []

    themes_out = []
    under, crowded = [], []
    for th in lm["themes"]:
        th_tokens = tokens(" ".join(th["keywords"]))
        on = sorted(
            a.get("id") for a in alts
            if tokens(a.get("label", "") + " " + " ".join(_as_list(a.get("key_constructs")))) & th_tokens
        )
        focal_on = bool(focal_tokens & th_tokens)
        n = len(on) + (1 if focal_on else 0)
        status = "under_theorised" if n == 0 else ("crowded" if n >= 2 else "covered")
        themes_out.append({
            "id": th["id"], "keywords": th["keywords"],
            "alternatives": on, "focal": focal_on, "status": status,
        })
        if status == "under_theorised":
            under.append(th["id"])
        elif status == "crowded":
            crowded.append(th["id"])

    return {
        "theory_id": T.get("id", ""),
        "themes": themes_out,
        "under_theorised_fronts": under,
        "redundancy_risk": crowded,
    }

theoryforge.fetch_corpus(query, per_page=25, mailto=None)

Build a corpus from the OpenAlex API (network call).

This adapter is assistive. It depends on a live external service whose results change over time, so it sits outside the package's deterministic core. per_page must be between 1 and 200, the range OpenAlex accepts.

Source code in src/theoryforge/lit.py

def fetch_corpus(query: str, per_page: int = 25, mailto: str | None = None) -> dict:
    """Build a corpus from the OpenAlex API (network call).

    This adapter is assistive. It depends on a live external service whose results
    change over time, so it sits outside the package's deterministic core.
    ``per_page`` must be between 1 and 200, the range OpenAlex accepts.
    """
    import urllib.parse
    import urllib.request

    # Reject out-of-range page sizes here rather than passing them through for
    # OpenAlex to reject, and with the same message the R twin uses.
    if isinstance(per_page, bool) or not isinstance(per_page, int) or not 1 <= per_page <= 200:
        raise ValueError("per_page must be between 1 and 200")
    params = {"search": query, "per-page": str(per_page)}
    if mailto:
        params["mailto"] = mailto
    url = "https://api.openalex.org/works?" + urllib.parse.urlencode(params)
    with urllib.request.urlopen(url, timeout=30) as resp:  # noqa: S310 (documented external call)
        data = json.load(resp)

    records = []
    for w in data.get("results", []):
        # Null display_name entries must be dropped before the emptiness test:
        # a keywords list made only of nulls is non-empty as returned, which
        # would suppress the concepts fallback and leave the record with no
        # keywords at all. The R twin filters first for the same reason.
        kws = [k.get("display_name") for k in (w.get("keywords") or [])]
        kws = [k for k in kws if k]
        if not kws:
            kws = [c.get("display_name") for c in (w.get("concepts") or [])[:5]]
            kws = [k for k in kws if k]
        records.append({
            "id": w.get("id"),
            "title": w.get("title"),
            "year": w.get("publication_year"),
            "keywords": kws,
            "references": w.get("referenced_works") or [],
        })
    return {"schema_version": "1.0", "id": f"openalex:{query}", "records": records}

theoryforge.new_evidence_dois(theory, candidate_dois)

DOIs in candidate_dois not already cited by the theory's evidence or alternatives.

Compares by normalised form (lowercased, with any doi.org/dx.doi.org URL prefix stripped), so a fresh literature search, for example via OpenAlex, Scopus, or any other source, can be checked against what the theory already engages with. Returns the qualifying DOIs in their original form, deduplicated and sorted by normalised form. Deterministic and takes no network dependency: the search itself is left to whichever literature tool the caller prefers.

Source code in src/theoryforge/lit.py

def new_evidence_dois(theory, candidate_dois: list) -> list:
    """DOIs in `candidate_dois` not already cited by the theory's evidence or alternatives.

    Compares by normalised form (lowercased, with any doi.org/dx.doi.org URL prefix
    stripped), so a fresh literature search, for example via OpenAlex, Scopus, or any
    other source, can be checked against what the theory already engages with. Returns
    the qualifying DOIs in their original form, deduplicated and sorted by normalised
    form. Deterministic and takes no network dependency: the search itself is left to
    whichever literature tool the caller prefers.
    """
    T = theory.data if hasattr(theory, "data") else theory
    known = set()
    for e in T.get("evidence") or []:
        doi = e.get("source_doi")
        if doi:
            known.add(_normalize_doi(doi))
    for a in T.get("alternatives") or []:
        doi = a.get("source_doi")
        if doi:
            known.add(_normalize_doi(doi))

    seen, out = set(), []
    for doi in candidate_dois or []:
        if not doi:
            continue
        norm = _normalize_doi(doi)
        if norm in known or norm in seen:
            continue
        seen.add(norm)
        out.append(doi)
    return sorted(out, key=_normalize_doi)