causalatee.graph

See Causal Graphs for a guide and the CGF 1.0 specification for the on-disk format.

graph

Classes

CauseEffectEdge

CauseEffectEdge(graph: 'CauseEffectGraph', cause: CauseEffectNode, effect: CauseEffectNode, *, count: int, necessity: float, sufficiency: float)

Bases: Edge[CauseEffectNode, 'CauseEffectEdge']

A directed cause->effect edge with an occurrence count and two causal-strength scores.

Source code in causalatee/graph/_cause_effect_graph.py

def __init__(
    self,
    graph: "CauseEffectGraph",
    cause: CauseEffectNode,
    effect: CauseEffectNode,
    *,
    count: int,
    necessity: float,
    sufficiency: float,
) -> None:
    super().__init__(graph)
    self._cause = cause
    self._effect = effect
    self._count = count
    self._necessity = necessity
    self._sufficiency = sufficiency

Attributes

cause property

cause: CauseEffectNode

Return the cause node.

effect property

effect: CauseEffectNode

Return the effect node.

count property

count: int

Return the number of occurrences supporting this pair.

necessity property

necessity: float

Return the necessity causal-strength score.

sufficiency property

sufficiency: float

Return the sufficiency causal-strength score.

CauseEffectNode

CauseEffectNode(graph: 'CauseEffectGraph', token: str)

Bases: Node['CauseEffectNode', 'CauseEffectEdge']

A single lexical (word-level) node in the Cause-Effect Graph.

Source code in causalatee/graph/_cause_effect_graph.py

def __init__(self, graph: "CauseEffectGraph", token: str) -> None:
    super().__init__(graph)
    self._token = token

Attributes

token property

token: str

Return the lexical token this node represents.

CauseNetEdge

CauseNetEdge(graph: CauseNet, cause: CauseNetNode, effect: CauseNetNode, *, sources: Sequence[Mapping[str, object]], support: int)

Bases: Edge[CauseNetNode, 'CauseNetEdge']

A directed CauseNet relation from a cause to an effect.

Source code in causalatee/graph/_causenet.py

def __init__(
    self,
    graph: CauseNet,
    cause: CauseNetNode,
    effect: CauseNetNode,
    *,
    sources: Sequence[Mapping[str, object]],
    support: int,
) -> None:
    super().__init__(graph)
    self._cause = cause
    self._effect = effect
    self._sources = tuple(sources)
    self._support = support

Attributes

cause property

cause: CauseNetNode

Return the cause node.

effect property

effect: CauseNetNode

Return the effect node.

sources property

sources: tuple[Mapping[str, object], ...]

Return the source records supporting this relation.

support property

support: int

Return the number of source records supporting this relation.

CauseNetNode

CauseNetNode(graph: CauseNet, concept: str)

Bases: Node['CauseNetNode', 'CauseNetEdge']

A CauseNet concept node.

Source code in causalatee/graph/_causenet.py

def __init__(self, graph: CauseNet, concept: str) -> None:
    super().__init__(graph)
    self._concept = concept

Attributes

concept property

concept: str

Return the CauseNet concept represented by this node.

Edge

Edge(graph: Graph[N, E])

Bases: Generic[N, E], ABC

A directed edge associated with a graph backend.

Source code in causalatee/graph/_graph.py

def __init__(self, graph: Graph[N, E]) -> None:
    self._graph = graph

Attributes

graph property

graph: Graph[N, E]

Return the graph that owns this edge.

source abstractmethod property

source: N

Return the source node.

target abstractmethod property

target: N

Return the target node.

metadata property

metadata: Mapping[str, object]

Return backend-provided metadata for this edge.

Graph

Bases: Generic[N, E], ABC

Abstract interface for a directed graph with typed nodes and edges.

Attributes

nodes abstractmethod property

nodes: Collection[N]

Return a collection view of the graph's nodes.

edges abstractmethod property

edges: Collection[E]

Return a collection view of the graph's edges.

node_metadata_schema property

node_metadata_schema: Mapping[str, object] | str | None

Return the Avro writer schema for node metadata, if defined.

edge_metadata_schema property

edge_metadata_schema: Mapping[str, object] | str | None

Return the Avro writer schema for edge metadata, if defined.

Methods:

get_node abstractmethod

get_node(node_id: str) -> N

Return the node identified by node_id.

Implementations should raise KeyError when no node exists.

Source code in causalatee/graph/_graph.py

@abstractmethod
def get_node(self, node_id: str) -> N:
    """Return the node identified by ``node_id``.

    Implementations should raise ``KeyError`` when no node exists.
    """

    raise NotImplementedError

edges_from abstractmethod

edges_from(node: N) -> Iterable[E]

Iterate over edges whose source is node.

Source code in causalatee/graph/_graph.py

@abstractmethod
def edges_from(self, node: N) -> Iterable[E]:
    """Iterate over edges whose source is ``node``."""

    raise NotImplementedError

edges_to abstractmethod

edges_to(node: N) -> Iterable[E]

Iterate over edges whose target is node.

Source code in causalatee/graph/_graph.py

@abstractmethod
def edges_to(self, node: N) -> Iterable[E]:
    """Iterate over edges whose target is ``node``."""

    raise NotImplementedError

node_metadata

node_metadata(node: N) -> Mapping[str, object]

Return resource-specific node metadata.

Source code in causalatee/graph/_graph.py

def node_metadata(self, node: N) -> Mapping[str, object]:
    """Return resource-specific node metadata."""

    return {}

edge_metadata

edge_metadata(edge: E) -> Mapping[str, object]

Return resource-specific edge metadata.

Source code in causalatee/graph/_graph.py

def edge_metadata(self, edge: E) -> Mapping[str, object]:
    """Return resource-specific edge metadata."""

    return {}

Node

Node(graph: Graph[N, E])

Bases: Generic[N, E], ABC

A lightweight node associated with a graph backend.

Source code in causalatee/graph/_graph.py

def __init__(self, graph: Graph[N, E]) -> None:
    self._graph = graph

Attributes

graph property

graph: Graph[N, E]

Return the graph that owns this node.

id abstractmethod property

id: str

Return the node identifier, unique within its graph.

metadata property

metadata: Mapping[str, object]

Return backend-provided metadata for this node.

Methods:

outgoing_edges

outgoing_edges() -> Iterable[E]

Iterate over edges whose source is this node.

Source code in causalatee/graph/_graph.py

def outgoing_edges(self: N) -> Iterable[E]:
    """Iterate over edges whose source is this node."""

    return self._graph.edges_from(self)

incoming_edges

incoming_edges() -> Iterable[E]

Iterate over edges whose target is this node.

Source code in causalatee/graph/_graph.py

def incoming_edges(self: N) -> Iterable[E]:
    """Iterate over edges whose target is this node."""

    return self._graph.edges_to(self)

SQLGraph

SQLGraph(connection: Connection)

Bases: Graph[SQLGraphNode, SQLGraphEdge]

A mutable graph backed by a caller-supplied SQLite connection. See module docstring.

Source code in causalatee/graph/_sql_graph.py

def __init__(self, connection: sqlite3.Connection) -> None:
    self._connection = connection
    self._connection.execute("CREATE TABLE IF NOT EXISTS nodes (id TEXT PRIMARY KEY, metadata TEXT NOT NULL)")
    self._connection.execute(
        "CREATE TABLE IF NOT EXISTS edges (source TEXT NOT NULL, target TEXT NOT NULL, metadata TEXT NOT NULL)"
    )
    self._connection.execute("CREATE INDEX IF NOT EXISTS sqlgraph_edges_source_idx ON edges (source)")
    self._connection.execute("CREATE INDEX IF NOT EXISTS sqlgraph_edges_target_idx ON edges (target)")
    self._connection.commit()

Methods:

add_node

add_node(node_id: str, metadata: Mapping[str, object] | None = None) -> SQLGraphNode

Add (or replace) a node. Safe to call again for an existing id -- overwrites its metadata.

Source code in causalatee/graph/_sql_graph.py

def add_node(self, node_id: str, metadata: Mapping[str, object] | None = None) -> SQLGraphNode:
    """Add (or replace) a node. Safe to call again for an existing id -- overwrites its metadata."""

    payload = dict(metadata or {})
    self._connection.execute(
        "INSERT OR REPLACE INTO nodes (id, metadata) VALUES (?, ?)", (node_id, json.dumps(payload))
    )
    self._connection.commit()
    return SQLGraphNode(self, node_id, payload)

add_edge

add_edge(source_id: str, target_id: str, metadata: Mapping[str, object] | None = None) -> SQLGraphEdge

Add a directed edge, auto-creating either endpoint (with empty metadata) if it doesn't exist yet -- same convention networkx's add_edge uses. Does NOT deduplicate or merge with an existing edge between the same pair (a multigraph, in effect); callers that need one aggregated edge per pair must aggregate before calling this once per pair (see causalatee.mining.GraphSink).

Source code in causalatee/graph/_sql_graph.py

def add_edge(self, source_id: str, target_id: str, metadata: Mapping[str, object] | None = None) -> SQLGraphEdge:
    """Add a directed edge, auto-creating either endpoint (with empty metadata) if it doesn't exist yet
    -- same convention networkx's ``add_edge`` uses. Does NOT deduplicate or merge with an existing
    edge between the same pair (a multigraph, in effect); callers that need one aggregated edge per
    pair must aggregate before calling this once per pair (see ``causalatee.mining.GraphSink``)."""

    for node_id in (source_id, target_id):
        self._connection.execute("INSERT OR IGNORE INTO nodes (id, metadata) VALUES (?, '{}')", (node_id,))
    payload = dict(metadata or {})
    self._connection.execute(
        "INSERT INTO edges (source, target, metadata) VALUES (?, ?, ?)",
        (source_id, target_id, json.dumps(payload)),
    )
    self._connection.commit()
    return SQLGraphEdge(self, self.get_node(source_id), self.get_node(target_id), payload)

SQLGraphEdge

SQLGraphEdge(graph: SQLGraph, source: SQLGraphNode, target: SQLGraphNode, metadata: Mapping[str, object])

Bases: Edge[SQLGraphNode, 'SQLGraphEdge']

A directed edge in a SQLGraph. Metadata is fetched once at construction time, not re-queried on access.

Source code in causalatee/graph/_sql_graph.py

def __init__(
    self, graph: SQLGraph, source: SQLGraphNode, target: SQLGraphNode, metadata: Mapping[str, object]
) -> None:
    super().__init__(graph)
    self._source = source
    self._target = target
    self._metadata = metadata

SQLGraphNode

SQLGraphNode(graph: SQLGraph, node_id: str, metadata: Mapping[str, object])

Bases: Node['SQLGraphNode', 'SQLGraphEdge']

A node in a SQLGraph. Metadata is fetched once at construction time, not re-queried on access.

Source code in causalatee/graph/_sql_graph.py

def __init__(self, graph: SQLGraph, node_id: str, metadata: Mapping[str, object]) -> None:
    super().__init__(graph)
    self._id = node_id
    self._metadata = metadata

Functions:

load_cause_effect_graph

load_cause_effect_graph(source: Source, *, limit: int | None = None) -> CauseEffectGraph

Load a Lexical Cause-Effect Graph (CEG) file from source.

source must be a local path to the tab-separated CEG text file (see module docstring for the exact line format).

Unlike load_causenet, omitting limit is safe even for the complete, tens-of-millions-of-edges release: node vocabulary is materialized eagerly (cheap -- a real release has on the order of 10^4-10^5 unique tokens), but edges are streamed from disk rather than materialized as Python objects, so save_cgf(load_cause_effect_graph(path), out_path) converts a file far larger than available RAM.

limit, when given, caps the number of EDGE LINES read from the top of the file -- applied consistently to both the node vocabulary built here and every later .edges iteration -- for quick examples/tests, exactly like load_causenet's own limit.

Source code in causalatee/graph/_cause_effect_graph.py

def load_cause_effect_graph(source: Source, *, limit: int | None = None) -> CauseEffectGraph:
    """Load a Lexical Cause-Effect Graph (CEG) file from ``source``.

    ``source`` must be a local path to the tab-separated CEG text file (see module docstring for the exact line
    format).

    Unlike ``load_causenet``, omitting ``limit`` is safe even for the complete, tens-of-millions-of-edges release:
    node vocabulary is materialized eagerly (cheap -- a real release has on the order of 10^4-10^5 unique tokens),
    but edges are streamed from disk rather than materialized as Python objects, so
    ``save_cgf(load_cause_effect_graph(path), out_path)`` converts a file far larger than available RAM.

    ``limit``, when given, caps the number of EDGE LINES read from the top of the file -- applied consistently to
    both the node vocabulary built here and every later ``.edges`` iteration -- for quick examples/tests, exactly
    like ``load_causenet``'s own ``limit``.
    """

    return CauseEffectGraph(source, limit=limit)

load_causenet

load_causenet(source: Source, *, limit: int | None = None, timeout: float = 60.0) -> CauseNet

Load CauseNet from source.

Omit limit to load the complete resource. This eager implementation may use substantial memory for the full graph.

Source code in causalatee/graph/_causenet.py

def load_causenet(
    source: Source,
    *,
    limit: int | None = None,
    timeout: float = 60.0,
) -> CauseNet:
    """Load CauseNet from ``source``.

    Omit ``limit`` to load the complete resource.  This eager implementation
    may use substantial memory for the full graph.
    """

    return CauseNet(source, limit=limit, timeout=timeout)

diff_avro_schema

diff_avro_schema(expected: object, actual: object, *, path: str = '$') -> list[str]

Recursively compare two Avro (sub-)schemas -- e.g. two infer_avro_schema results, or two field type values nested within them -- and return every structural difference found (missing fields, unexpected extra fields, type mismatches) as a human-readable string. An empty list means the two schemas describe the same shape.

Only type/fields/items are compared -- a record's own name/namespace are deterministic byproducts of field path (see infer_avro_schema), not part of its shape, so two schemas naming the same nested record differently are NOT reported as different here. Field order doesn't matter either, only presence and type.

path accumulates a location for each reported difference (e.g. $.evidence[].topic) so a caller can point at exactly where two metadata shapes disagree, rather than just reporting "they differ".

Source code in causalatee/graph/_cgf.py

def diff_avro_schema(expected: object, actual: object, *, path: str = "$") -> list[str]:
    """Recursively compare two Avro (sub-)schemas -- e.g. two [`infer_avro_schema`][causalatee.graph.infer_avro_schema]
    results, or two field ``type`` values nested within them -- and return every structural difference found
    (missing fields, unexpected extra fields, type mismatches) as a human-readable string. An empty list means
    the two schemas describe the same shape.

    Only ``type``/``fields``/``items`` are compared -- a record's own ``name``/``namespace`` are deterministic
    byproducts of field path (see ``infer_avro_schema``), not part of its shape, so two schemas naming the same
    nested record differently are NOT reported as different here. Field order doesn't matter either, only
    presence and type.

    ``path`` accumulates a location for each reported difference (e.g. ``$.evidence[].topic``) so a caller can
    point at exactly where two metadata shapes disagree, rather than just reporting "they differ".
    """

    expected_kind = expected.get("type") if isinstance(expected, Mapping) else expected
    actual_kind = actual.get("type") if isinstance(actual, Mapping) else actual

    if expected_kind != actual_kind:
        return [f"{path}: expected type {expected_kind!r}, got {actual_kind!r}"]

    if expected_kind == "record":
        assert isinstance(expected, Mapping) and isinstance(actual, Mapping)  # guaranteed by the type check above
        expected_fields = {field["name"]: field["type"] for field in expected.get("fields", [])}
        actual_fields = {field["name"]: field["type"] for field in actual.get("fields", [])}
        missing = sorted(expected_fields.keys() - actual_fields.keys())
        extra = sorted(actual_fields.keys() - expected_fields.keys())
        diffs = [f"{path}.{name}: missing" for name in missing]
        diffs += [f"{path}.{name}: unexpected extra field" for name in extra]
        for name in sorted(expected_fields.keys() & actual_fields.keys()):
            diffs.extend(diff_avro_schema(expected_fields[name], actual_fields[name], path=f"{path}.{name}"))
        return diffs

    if expected_kind == "array":
        assert isinstance(expected, Mapping) and isinstance(actual, Mapping)  # guaranteed by the type check above
        return diff_avro_schema(expected.get("items"), actual.get("items"), path=f"{path}[]")

    return []

infer_avro_schema

infer_avro_schema(name: str, example: Mapping[str, object], *, namespace: str = 'causalatee.inferred') -> dict

Infer an Avro record schema from one example node.metadata/edge.metadata object.

Only handles "plain data" values -- str, bool, int, float, None, list/tuple, and nested Mapping -- exactly the shapes JSON-like metadata is built from. Raises :class:TypeError for anything else (a custom class instance, a set, ...), since there's no principled way to guess an Avro type for it from a single example.

Field order in example becomes field order in the returned schema (Python dicts preserve insertion order, and so does Avro's own fields list) -- deterministic for a given example. Nested records (dicts, and dicts nested inside list items) get their own generated, path-qualified name (e.g. "EdgeMetadata.Evidence.Item") so two structurally different fields that happen to share a key name (e.g. a top-level "source" and a "source" nested inside each "evidence" entry) never collide under Avro's global name-uniqueness rule.

LIMITATION, stated plainly rather than hidden: this infers from exactly ONE example. A field that is sometimes a different type across the real population, or a list that happens to be empty in this particular example (inferred as a permissive nullable-string array, since there's nothing to inspect), won't be caught here -- this assumes every value that will actually be encoded against the returned schema shares this one example's shape, which holds for one caller using one consistent metadata shape (e.g. one graph_sink() run over one corpus), not a general-purpose schema-drift detector across a heterogeneous population.

Source code in causalatee/graph/_cgf.py

def infer_avro_schema(name: str, example: Mapping[str, object], *, namespace: str = "causalatee.inferred") -> dict:
    """Infer an Avro record schema from one example ``node.metadata``/``edge.metadata`` object.

    Only handles "plain data" values -- ``str``, ``bool``, ``int``, ``float``, ``None``, ``list``/``tuple``, and
    nested ``Mapping`` -- exactly the shapes JSON-like metadata is built from. Raises :class:`TypeError` for
    anything else (a custom class instance, a set, ...), since there's no principled way to guess an Avro type for
    it from a single example.

    Field order in ``example`` becomes field order in the returned schema (Python dicts preserve insertion order,
    and so does Avro's own ``fields`` list) -- deterministic for a given example. Nested records (dicts, and dicts
    nested inside list items) get their own generated, path-qualified name (e.g. ``"EdgeMetadata.Evidence.Item"``)
    so two structurally different fields that happen to share a key name (e.g. a top-level ``"source"`` and a
    ``"source"`` nested inside each ``"evidence"`` entry) never collide under Avro's global name-uniqueness rule.

    LIMITATION, stated plainly rather than hidden: this infers from exactly ONE example. A field that is sometimes
    a different type across the real population, or a list that happens to be empty in this particular example
    (inferred as a permissive nullable-string array, since there's nothing to inspect), won't be caught here --
    this assumes every value that will actually be encoded against the returned schema shares this one example's
    shape, which holds for one caller using one consistent metadata shape (e.g. one ``graph_sink()`` run over one
    corpus), not a general-purpose schema-drift detector across a heterogeneous population.
    """

    return {
        "type": "record",
        "name": name,
        "namespace": namespace,
        "fields": [
            {"name": key, "type": _infer_avro_type(key, value, path=f"{namespace}.{name}")}
            for key, value in example.items()
        ],
    }

load_cgf

load_cgf(path: PathLike, *, validate: bool = False) -> CGFGraph

Memory-map and return a read-only CGF graph.

Source code in causalatee/graph/_cgf.py

def load_cgf(path: PathLike, *, validate: bool = False) -> CGFGraph:
    """Memory-map and return a read-only CGF graph."""

    return CGFGraph(path, validate=validate)

save_cgf

save_cgf(graph: Graph[Any, Any], path: PathLike, *, include_incoming: bool = True, node_metadata_schema: Mapping[str, object] | str | None = None, edge_metadata_schema: Mapping[str, object] | str | None = None) -> None

Serialize graph using disk-backed sorting and streamed sections.

Memory use is bounded independently of metadata and edge count. SQLite is used as a disposable external-sort spool; final CGF sections are written to separate temporary streams and assembled atomically. The only graph-side state retained by this function is whatever the supplied graph itself owns.

Source code in causalatee/graph/_cgf.py

def save_cgf(
    graph: Graph[Any, Any],
    path: PathLike,
    *,
    include_incoming: bool = True,
    node_metadata_schema: Mapping[str, object] | str | None = None,
    edge_metadata_schema: Mapping[str, object] | str | None = None,
) -> None:
    """Serialize ``graph`` using disk-backed sorting and streamed sections.

    Memory use is bounded independently of metadata and edge count. SQLite is
    used as a disposable external-sort spool; final CGF sections are written to
    separate temporary streams and assembled atomically. The only graph-side
    state retained by this function is whatever the supplied graph itself owns.
    """

    destination = Path(path)
    destination.parent.mkdir(parents=True, exist_ok=True)
    schemas, schema_payloads, parsed_schemas = _prepare_writer_schemas(
        graph,
        node_metadata_schema=node_metadata_schema,
        edge_metadata_schema=edge_metadata_schema,
    )

    with tempfile.TemporaryDirectory(prefix=f".{destination.name}.cgf-", dir=destination.parent) as temporary_directory:
        staging = Path(temporary_directory)
        database = sqlite3.connect(staging / "sort.sqlite3")
        try:
            _configure_spool(database)
            database.executescript(
                """
                CREATE TABLE nodes (
                    id BLOB PRIMARY KEY,
                    metadata BLOB NOT NULL,
                    ordinal INTEGER UNIQUE
                ) WITHOUT ROWID;
                CREATE TABLE edges (
                    source INTEGER NOT NULL,
                    target INTEGER NOT NULL,
                    input_order INTEGER NOT NULL,
                    metadata BLOB NOT NULL,
                    edge_index INTEGER
                );
                """
            )

            node_count = _spool_nodes(database, graph, parsed_schemas["node"], schemas["node"])
            sections = _emit_node_sections(database, staging, node_count, schema_payloads["node"])
            edge_count = _spool_edges(database, graph, parsed_schemas["edge"], schemas["edge"])
            sections.extend(
                _emit_outgoing_sections(
                    database,
                    staging,
                    node_count=node_count,
                    edge_count=edge_count,
                    edge_schema_payload=schema_payloads["edge"],
                )
            )

            flags = 0
            if include_incoming:
                flags |= FLAG_HAS_INCOMING
                sections.extend(
                    _emit_incoming_sections(
                        database,
                        staging,
                        node_count=node_count,
                        edge_count=edge_count,
                    )
                )
        finally:
            database.close()

        _write_staged_file(
            destination,
            sections,
            flags=flags,
            node_count=node_count,
            edge_count=edge_count,
        )