Python API Reference

Top-level exports

import polaroid

polaroid

polaroid - Embeddable CRDT scene graph for embodied AI agents.

ClosedLoopError

Bases: ValueError

Raised when the scene gate refuses empty or detached structure.

GateOutcome(ok, verdict, reason, exit_code, node_count=0, edge_count=0, detached_count=0, orphan_room_count=0) dataclass

Result of a closed-loop scene graph read.

Attributes:

Name	Type	Description
ok	bool	True only when navigation/attachment may proceed.
verdict	str	PASS, FAIL, or FAIL_LOUD.
reason	str	Always non-empty.
exit_code	int	0 PASS, 1 FAIL, 2 FAIL_LOUD.
node_count	int	Nodes examined.
edge_count	int	Edges examined.
detached_count	int	Nodes missing required attachment.
orphan_room_count	int	Rooms with no nav edges.

MergeResult(added_nodes, updated_nodes, added_edges, conflicts_resolved) dataclass

Result of merging two scene graphs.

to_dict()

Serialize to a plain dict.

Source code in src/polaroid/graph.py

def to_dict(self) -> dict:  # type: ignore[type-arg]
    """Serialize to a plain dict."""
    return {
        "added_nodes": [n.to_dict() for n in self.added_nodes],
        "updated_nodes": [n.to_dict() for n in self.updated_nodes],
        "added_edges": [e.to_dict() for e in self.added_edges],
        "conflicts_resolved": self.conflicts_resolved,
    }

summary()

Return a one-line summary string.

Source code in src/polaroid/graph.py

def summary(self) -> str:
    """Return a one-line summary string."""
    return (
        f"Added {len(self.added_nodes)} nodes, "
        f"updated {len(self.updated_nodes)} nodes, "
        f"added {len(self.added_edges)} edges, "
        f"resolved {self.conflicts_resolved} conflict(s)."
    )

SceneEdge(source_id, target_id, relation, confidence=1.0, observed_at=time.time()) dataclass

A directed spatial relationship between two nodes.

to_dict()

Serialize to a plain dict.

Source code in src/polaroid/graph.py

def to_dict(self) -> dict:  # type: ignore[type-arg]
    """Serialize to a plain dict."""
    return {
        "id": self.id,
        "source_id": self.source_id,
        "target_id": self.target_id,
        "relation": self.relation,
        "confidence": self.confidence,
        "observed_at": self.observed_at,
    }

from_dict(d) classmethod

Deserialize from a plain dict.

Source code in src/polaroid/graph.py

@classmethod
def from_dict(cls, d: dict) -> SceneEdge:  # type: ignore[type-arg]
    """Deserialize from a plain dict."""
    edge = cls(
        source_id=d["source_id"],
        target_id=d["target_id"],
        relation=d["relation"],
        confidence=d.get("confidence", 1.0),
        observed_at=d.get("observed_at", time.time()),
    )
    return edge

SceneNode(label, node_type, properties, confidence=1.0, observed_at=time.time(), agent_id='') dataclass

A node in the scene graph (object, room, surface, region, or agent).

to_dict()

Serialize to a plain dict.

Source code in src/polaroid/graph.py

def to_dict(self) -> dict:  # type: ignore[type-arg]
    """Serialize to a plain dict."""
    return {
        "id": self.id,
        "label": self.label,
        "node_type": self.node_type,
        "properties": self.properties,
        "confidence": self.confidence,
        "observed_at": self.observed_at,
        "agent_id": self.agent_id,
    }

from_dict(d) classmethod

Deserialize from a plain dict.

Source code in src/polaroid/graph.py

@classmethod
def from_dict(cls, d: dict) -> SceneNode:  # type: ignore[type-arg]
    """Deserialize from a plain dict."""
    node = cls(
        label=d["label"],
        node_type=d["node_type"],
        properties=d.get("properties", {}),
        confidence=d.get("confidence", 1.0),
        observed_at=d.get("observed_at", time.time()),
        agent_id=d.get("agent_id", ""),
    )
    return node

SceneMerger

Merge two SceneStore instances using CRDT semantics.

This is a pure function wrapped in a class for extensibility. Neither store is modified; the result is applied to local.

merge(local, remote)

Merge remote into local.

CRDT properties guaranteed: - Idempotent: merging the same remote twice produces the same result. - Commutative: merge(A, B) and merge(B, A) produce the same final state. - Associative: merge order among multiple remotes does not matter.

Parameters:

Name	Type	Description	Default
local	SceneStore	The destination SceneStore (written to).	required
remote	SceneStore	The source SceneStore (read only).	required

Returns:

Type	Description
MergeResult	MergeResult describing what changed.

Source code in src/polaroid/merger.py

def merge(self, local: SceneStore, remote: SceneStore) -> MergeResult:
    """Merge remote into local.

    CRDT properties guaranteed:
    - Idempotent: merging the same remote twice produces the same result.
    - Commutative: merge(A, B) and merge(B, A) produce the same final state.
    - Associative: merge order among multiple remotes does not matter.

    Args:
        local:  The destination SceneStore (written to).
        remote: The source SceneStore (read only).

    Returns:
        MergeResult describing what changed.
    """
    added_nodes: list[SceneNode] = []
    updated_nodes: list[SceneNode] = []
    added_edges: list[SceneEdge] = []
    conflicts_resolved = 0

    # ── Nodes ──────────────────────────────────────────────────────────────
    for remote_node in remote.list_nodes():
        local_node = local.get_node(remote_node.id)

        if local_node is None:
            # Grow-only set: new node, always add
            local.upsert_node(remote_node)
            added_nodes.append(remote_node)

        elif remote_node.confidence > local_node.confidence:
            # Confidence-weighted last-write-wins register
            local.upsert_node(remote_node)
            updated_nodes.append(remote_node)
            conflicts_resolved += 1

        # else: local has higher or equal confidence - keep local, no action

    # ── Edges ──────────────────────────────────────────────────────────────
    for remote_edge in remote.list_edges():
        local_edge = local.get_edge(remote_edge.id)

        if local_edge is None:
            local.upsert_edge(remote_edge)
            added_edges.append(remote_edge)

        elif remote_edge.confidence > local_edge.confidence:
            local.upsert_edge(remote_edge)
            conflicts_resolved += 1

    return MergeResult(
        added_nodes=added_nodes,
        updated_nodes=updated_nodes,
        added_edges=added_edges,
        conflicts_resolved=conflicts_resolved,
    )

SceneQuery(store)

High-level query interface for a SceneStore.

All methods are read-only - they never modify the store.

Source code in src/polaroid/query.py

def __init__(self, store: SceneStore) -> None:
    self._store = store

find_nodes(node_type=None, label_contains=None, min_confidence=0.0)

Return nodes matching the given filters.

Parameters:

Name	Type	Description	Default
node_type	str | None	Only return nodes of this type (e.g. "object", "room").	None
label_contains	str | None	Case-insensitive substring match on node label.	None
min_confidence	float	Exclude nodes with confidence below this threshold.	0.0

Returns:

Type	Description
list[SceneNode]	List of matching SceneNode objects.

Source code in src/polaroid/query.py

def find_nodes(
    self,
    node_type: str | None = None,
    label_contains: str | None = None,
    min_confidence: float = 0.0,
) -> list[SceneNode]:
    """Return nodes matching the given filters.

    Args:
        node_type:      Only return nodes of this type (e.g. "object", "room").
        label_contains: Case-insensitive substring match on node label.
        min_confidence: Exclude nodes with confidence below this threshold.

    Returns:
        List of matching SceneNode objects.
    """
    nodes = self._store.list_nodes(node_type=node_type, min_confidence=min_confidence)
    if label_contains is not None:
        needle = label_contains.lower()
        nodes = [n for n in nodes if needle in n.label.lower()]
    return nodes

find_neighbors(node_id, relation=None)

Return nodes that are targets of edges originating from node_id.

Parameters:

Name	Type	Description	Default
node_id	str	Source node ID.	required
relation	str | None	If given, only follow edges with this relation type.	None

Returns:

Type	Description
list[SceneNode]	List of neighbor SceneNode objects (may be empty).

Source code in src/polaroid/query.py

def find_neighbors(
    self,
    node_id: str,
    relation: str | None = None,
) -> list[SceneNode]:
    """Return nodes that are targets of edges originating from node_id.

    Args:
        node_id:  Source node ID.
        relation: If given, only follow edges with this relation type.

    Returns:
        List of neighbor SceneNode objects (may be empty).
    """
    edges = self._store.list_edges(source_id=node_id, relation=relation)
    neighbors: list[SceneNode] = []
    for edge in edges:
        target = self._store.get_node(edge.target_id)
        if target is not None:
            neighbors.append(target)
    return neighbors

context_summary(agent_id='')

Return a human-readable description of the scene.

Parameters:

Name	Type	Description	Default
agent_id	str	If non-empty, only count nodes observed by this agent.	''

Returns:

Type	Description
str	A one-paragraph text summary of the scene graph.

Source code in src/polaroid/query.py

def context_summary(self, agent_id: str = "") -> str:
    """Return a human-readable description of the scene.

    Args:
        agent_id: If non-empty, only count nodes observed by this agent.

    Returns:
        A one-paragraph text summary of the scene graph.
    """
    all_nodes = self._store.list_nodes()
    if agent_id:
        all_nodes = [n for n in all_nodes if n.agent_id == agent_id]

    # Count by type
    counts: dict[str, int] = {}
    for node in all_nodes:
        counts[node.node_type] = counts.get(node.node_type, 0) + 1

    # Build header
    type_parts = []
    for t in sorted(counts):
        c = counts[t]
        type_parts.append(f"{c} {t}{'s' if c != 1 else ''}")

    if not type_parts:
        return "Empty scene graph - no nodes observed."

    header = ", ".join(type_parts) + "."

    # Known objects preview
    object_labels = [n.label for n in all_nodes if n.node_type == "object"][:10]
    edge_count = self._store.edge_count()

    parts = [header]
    if object_labels:
        parts.append(f"Known objects: {', '.join(object_labels)}.")
    if edge_count > 0:
        suffix = "s" if edge_count != 1 else ""
        parts.append(f"{edge_count} spatial relationship{suffix} recorded.")

    return " ".join(parts)

GraphStats(node_count, edge_count, node_types, edge_relations, avg_degree, max_degree, connected_components, diameter) dataclass

Statistics about a scene graph.

SceneStore(path, *, data_root=None)

SQLite-backed persistent store for scene nodes and edges.

All writes are immediately committed. One SceneStore per process.

Paths are confined to a data root (default .polaroid or POLAROID_DATA_DIR) to prevent path-injection from API/MCP callers (CWE-22 / CodeQL py/path-injection).

Open or create a store at path.

Parameters

path: Store path or ":memory:". Relative paths are resolved under data_root. Absolute paths must stay inside data_root. data_root: Optional root directory (defaults to env POLAROID_DATA_DIR or .polaroid). Tests should pass the pytest tmp_path.

Source code in src/polaroid/store.py

def __init__(self, path: str | Path, *, data_root: str | Path | None = None) -> None:
    """Open or create a store at *path*.

    Parameters
    ----------
    path:
        Store path or ``":memory:"``. Relative paths are resolved under
        *data_root*. Absolute paths must stay inside *data_root*.
    data_root:
        Optional root directory (defaults to env ``POLAROID_DATA_DIR`` or
        ``.polaroid``). Tests should pass the pytest ``tmp_path``.
    """
    resolved = safe_db_path(path, root=data_root)
    if resolved == MEMORY_URI:
        self._path = Path(MEMORY_URI)
        self._conn = sqlite3.connect(MEMORY_URI)
    else:
        self._path = Path(resolved)
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._conn = sqlite3.connect(str(self._path))
    self._conn.row_factory = sqlite3.Row
    self._create_schema()

upsert_node(node)

Insert or update node - only overwrites if incoming confidence >= existing.

Source code in src/polaroid/store.py

def upsert_node(self, node: SceneNode) -> None:
    """Insert or update node - only overwrites if incoming confidence >= existing."""
    existing = self.get_node(node.id)
    if existing is None:
        self._conn.execute(
            """
            INSERT INTO nodes
                (id, label, node_type, properties, confidence, observed_at, agent_id)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                node.id,
                node.label,
                node.node_type,
                json.dumps(node.properties),
                node.confidence,
                node.observed_at,
                node.agent_id,
            ),
        )
    elif node.confidence >= existing.confidence:
        self._conn.execute(
            """
            UPDATE nodes SET properties=?, confidence=?, observed_at=?, agent_id=?
            WHERE id=?
            """,
            (
                json.dumps(node.properties),
                node.confidence,
                node.observed_at,
                node.agent_id,
                node.id,
            ),
        )
    self._conn.commit()

get_node(node_id)

Return a SceneNode by ID, or None if not found.

Source code in src/polaroid/store.py

def get_node(self, node_id: str) -> SceneNode | None:
    """Return a SceneNode by ID, or None if not found."""
    row = self._conn.execute("SELECT * FROM nodes WHERE id=?", (node_id,)).fetchone()
    if row is None:
        return None
    return self._row_to_node(row)

list_nodes(node_type=None, min_confidence=0.0)

Return nodes, optionally filtered by type and min confidence.

Source code in src/polaroid/store.py

def list_nodes(
    self,
    node_type: str | None = None,
    min_confidence: float = 0.0,
) -> list[SceneNode]:
    """Return nodes, optionally filtered by type and min confidence."""
    sql = "SELECT * FROM nodes WHERE confidence >= ?"
    params: list[Any] = [min_confidence]
    if node_type is not None:
        sql += " AND node_type=?"
        params.append(node_type)
    rows = self._conn.execute(sql, params).fetchall()
    return [self._row_to_node(r) for r in rows]

upsert_edge(edge)

Insert or update edge - only overwrites if incoming confidence >= existing.

Source code in src/polaroid/store.py

def upsert_edge(self, edge: SceneEdge) -> None:
    """Insert or update edge - only overwrites if incoming confidence >= existing."""
    existing = self.get_edge(edge.id)
    if existing is None:
        self._conn.execute(
            """
            INSERT INTO edges (id, source_id, target_id, relation, confidence, observed_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                edge.id,
                edge.source_id,
                edge.target_id,
                edge.relation,
                edge.confidence,
                edge.observed_at,
            ),
        )
    elif edge.confidence >= existing.confidence:
        self._conn.execute(
            "UPDATE edges SET confidence=?, observed_at=? WHERE id=?",
            (edge.confidence, edge.observed_at, edge.id),
        )
    self._conn.commit()

get_edge(edge_id)

Return a SceneEdge by ID, or None if not found.

Source code in src/polaroid/store.py

def get_edge(self, edge_id: str) -> SceneEdge | None:
    """Return a SceneEdge by ID, or None if not found."""
    row = self._conn.execute("SELECT * FROM edges WHERE id=?", (edge_id,)).fetchone()
    if row is None:
        return None
    return self._row_to_edge(row)

list_edges(source_id=None, relation=None)

Return edges, optionally filtered by source_id and/or relation.

Source code in src/polaroid/store.py

def list_edges(
    self,
    source_id: str | None = None,
    relation: str | None = None,
) -> list[SceneEdge]:
    """Return edges, optionally filtered by source_id and/or relation."""
    sql = "SELECT * FROM edges WHERE 1=1"
    params: list[Any] = []
    if source_id is not None:
        sql += " AND source_id=?"
        params.append(source_id)
    if relation is not None:
        sql += " AND relation=?"
        params.append(relation)
    rows = self._conn.execute(sql, params).fetchall()
    return [self._row_to_edge(r) for r in rows]

node_count()

Return total number of nodes.

Source code in src/polaroid/store.py

def node_count(self) -> int:
    """Return total number of nodes."""
    row = self._conn.execute("SELECT COUNT(*) FROM nodes").fetchone()
    return int(row[0]) if row else 0

edge_count()

Return total number of edges.

Source code in src/polaroid/store.py

def edge_count(self) -> int:
    """Return total number of edges."""
    row = self._conn.execute("SELECT COUNT(*) FROM edges").fetchone()
    return int(row[0]) if row else 0

close()

Close the database connection.

Source code in src/polaroid/store.py

def close(self) -> None:
    """Close the database connection."""
    self._conn.close()

assert_line_of_sight(store, observer_id, target_id, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_line_of_sight is ok.

Source code in src/polaroid/closed_loop.py

def assert_line_of_sight(
    store: SceneStore,
    observer_id: str,
    target_id: str,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_line_of_sight` is ok."""
    outcome = gate_line_of_sight(store, observer_id, target_id, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_attachment(store, part_id, parent_id)

Gate a specific part→parent attachment claim (DETACHED-PART).

Source code in src/polaroid/closed_loop.py

def gate_attachment(
    store: SceneStore,
    part_id: str,
    parent_id: str,
) -> GateOutcome:
    """Gate a specific part→parent attachment claim (DETACHED-PART)."""
    n = store.node_count()
    e = store.edge_count()
    part = store.get_node(part_id)
    parent = store.get_node(parent_id)
    if part is None or parent is None:
        return _fail_loud(
            f"missing nodes for attachment part={part_id!r} parent={parent_id!r}",
            node_count=n,
            edge_count=e,
            detached_count=1,
        )
    if is_attached(store, part_id, parent_id):
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=f"attachment ok: {part.label!r} linked to {parent.label!r}",
            exit_code=0,
            node_count=n,
            edge_count=e,
        )
    return _fail(
        f"DETACHED-PART: {part.label!r} ({part_id[:8]}…) has no attachment edge "
        f"to {parent.label!r} - thruster/body weld missing",
        node_count=n,
        edge_count=e,
        detached_count=1,
    )

gate_line_of_sight(store, observer_id, target_id, *, action='observe', max_hops=8, require_nodes=True)

Refuse observe/target/interact when graph LOS is missing (LINE-OF-SIGHT).

Embodied agents often claim they can see or act on a target when walls / occluders intervene. gate_navigable only checks room connectivity; this gate checks visibility path observer→target.

Rules:

• Empty scene → FAIL_LOUD
• Missing observer/target nodes → FAIL_LOUD
• No LOS path → FAIL
• LOS present → PASS

Source code in src/polaroid/closed_loop.py

def gate_line_of_sight(
    store: SceneStore,
    observer_id: str,
    target_id: str,
    *,
    action: str = "observe",
    max_hops: int = 8,
    require_nodes: bool = True,
) -> GateOutcome:
    """Refuse observe/target/interact when graph LOS is missing (LINE-OF-SIGHT).

    Embodied agents often claim they can see or act on a target when walls /
    occluders intervene. ``gate_navigable`` only checks room connectivity;
    this gate checks **visibility path** observer→target.

    Rules:

    * Empty scene → **FAIL_LOUD**
    * Missing observer/target nodes → **FAIL_LOUD**
    * No LOS path → **FAIL**
    * LOS present → **PASS**
    """
    n, e = store.node_count(), store.edge_count()
    if n == 0:
        return _fail_loud(
            "LINE-OF-SIGHT: empty scene — cannot claim visibility on phantom map",
            node_count=0,
            edge_count=0,
        )

    obs = store.get_node(observer_id)
    tgt = store.get_node(target_id)
    if require_nodes and (obs is None or tgt is None):
        missing = []
        if obs is None:
            missing.append("observer")
        if tgt is None:
            missing.append("target")
        return _fail_loud(
            f"LINE-OF-SIGHT: missing nodes {missing} for action={action!r} "
            f"observer={observer_id[:8]}… target={target_id[:8]}…",
            node_count=n,
            edge_count=e,
        )

    if has_line_of_sight(store, observer_id, target_id, max_hops=max_hops):
        olab = obs.label if obs else observer_id[:8]
        tlab = tgt.label if tgt else target_id[:8]
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"LINE-OF-SIGHT ok: {olab!r} → {tlab!r} action={action!r} "
                f"hops<={max_hops}"
            ),
            exit_code=0,
            node_count=n,
            edge_count=e,
        )

    olab = obs.label if obs else observer_id[:8]
    tlab = tgt.label if tgt else target_id[:8]
    return _fail(
        f"LINE-OF-SIGHT: no visibility path {olab!r} → {tlab!r} for "
        f"action={action!r} — refuse observe/target through occlusion "
        f"(walls/blocks-view without sees edge)",
        node_count=n,
        edge_count=e,
    )

gate_navigable(store, *, min_rooms=2)

Gate multi-room navigation: rooms need adjacent/connects edges.

PRIMAL3 / multi-agent pathfinding class - disconnected rooms are not a map.

Source code in src/polaroid/closed_loop.py

def gate_navigable(
    store: SceneStore,
    *,
    min_rooms: int = 2,
) -> GateOutcome:
    """Gate multi-room navigation: rooms need adjacent/connects edges.

    PRIMAL3 / multi-agent pathfinding class - disconnected rooms are not a map.
    """
    nodes = store.list_nodes()
    rooms = [n for n in nodes if n.node_type.lower() in {"room", "region", "area"}]
    n, e = len(nodes), store.edge_count()
    if len(rooms) < min_rooms:
        # Not a multi-room map - pass navigable check (single space)
        if store.node_count() == 0:
            return _fail_loud("empty scene - not navigable", node_count=0, edge_count=0)
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=f"navigable: {len(rooms)} room(s) (<{min_rooms} multi-room threshold)",
            exit_code=0,
            node_count=n,
            edge_count=e,
        )

    nav = nav_edges(store)
    if not nav:
        return _fail(
            f"EMPTY-SCENE connectivity: {len(rooms)} rooms but 0 nav edges "
            f"(adjacent-to/connects) - pathfinding impossible",
            node_count=n,
            edge_count=e,
            orphan_room_count=len(rooms),
        )

    # rooms with no incident nav edge
    linked: set[str] = set()
    for edge in nav:
        linked.add(edge.source_id)
        linked.add(edge.target_id)
    orphans = [r for r in rooms if r.id not in linked]
    if orphans:
        return _fail(
            f"orphan rooms without nav edges: {[o.label for o in orphans[:5]]}",
            node_count=n,
            edge_count=e,
            orphan_room_count=len(orphans),
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"navigable: {len(rooms)} rooms, {len(nav)} nav edges",
        exit_code=0,
        node_count=n,
        edge_count=e,
        orphan_room_count=0,
    )

gate_scene(store, *, min_nodes=1, refuse_detached=True)

Gate a scene store: empty FAIL_LOUD; detached parts FAIL.

This is the load-bearing closed-loop reader for embodied agents.

Source code in src/polaroid/closed_loop.py

def gate_scene(
    store: SceneStore,
    *,
    min_nodes: int = 1,
    refuse_detached: bool = True,
) -> GateOutcome:
    """Gate a scene store: empty FAIL_LOUD; detached parts FAIL.

    This is the load-bearing closed-loop reader for embodied agents.
    """
    n = store.node_count()
    e = store.edge_count()
    if n < min_nodes:
        return _fail_loud(
            f"empty scene graph - {n} nodes (<{min_nodes}); "
            f"cannot navigate or merge a phantom map (EMPTY-SCENE)",
            node_count=n,
            edge_count=e,
        )

    if refuse_detached:
        det = detached_parts(store)
        if det:
            labels = [d.label for d in det[:5]]
            return _fail(
                f"DETACHED-PART: {len(det)} node(s) without attachment edges "
                f"(e.g. {labels}) - refuse silent weld (Roblox thruster class)",
                node_count=n,
                edge_count=e,
                detached_count=len(det),
            )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"scene ok: nodes={n} edges={e}",
        exit_code=0,
        node_count=n,
        edge_count=e,
        detached_count=0,
    )

has_line_of_sight(store, observer_id, target_id, *, max_hops=8)

True if observer can reach target via LOS/nav edges without occlusion cut.

Rules (graph-level LOS, not full raycast):

• Direct sees / visible-to edge observer→target → True
• BFS on undirected LOS+nav edges up to max_hops
• If an occludes edge names a node on the path as occluding the target, that path is rejected (simple: any occluder adjacent to target blocks unless a direct sees edge exists)

Source code in src/polaroid/closed_loop.py

def has_line_of_sight(
    store: SceneStore,
    observer_id: str,
    target_id: str,
    *,
    max_hops: int = 8,
) -> bool:
    """True if observer can reach target via LOS/nav edges without occlusion cut.

    Rules (graph-level LOS, not full raycast):

    * Direct ``sees`` / ``visible-to`` edge observer→target → True
    * BFS on undirected LOS+nav edges up to ``max_hops``
    * If an ``occludes`` edge names a node on the path as occluding the target,
      that path is rejected (simple: any occluder adjacent to target blocks
      unless a direct sees edge exists)
    """
    if observer_id == target_id:
        return True
    nodes = {n.id for n in store.list_nodes()}
    if observer_id not in nodes or target_id not in nodes:
        return False

    # Direct visibility wins over occlusion
    for e in store.list_edges():
        rel = _norm_rel(e.relation)
        if rel in {_norm_rel(r) for r in LOS_RELATIONS}:
            if e.source_id == observer_id and e.target_id == target_id:
                return True
            if e.source_id == target_id and e.target_id == observer_id:
                return True

    occ = _occlusion_pairs(store)
    # If target is occluded by something and no direct sees → may still walk
    # around via rooms unless occluder sits on all paths (approx: block if
    # occluder is neighbor of target and not the observer)
    blocked_neighbors = {a for a, b in occ if b == target_id}

    # Build adjacency from LOS + NAV
    adj: dict[str, set[str]] = defaultdict(set)
    for e in _los_edges(store):
        adj[e.source_id].add(e.target_id)
        adj[e.target_id].add(e.source_id)

    if observer_id not in adj and target_id not in adj:
        return False

    q: deque[tuple[str, int]] = deque([(observer_id, 0)])
    seen = {observer_id}
    while q:
        cur, dist = q.popleft()
        if cur == target_id:
            return True
        if dist >= max_hops:
            continue
        for nxt in adj.get(cur, ()):
            if nxt in seen:
                continue
            # do not step through known occluders of the target
            if nxt in blocked_neighbors and nxt != target_id:
                continue
            seen.add(nxt)
            q.append((nxt, dist + 1))
    return False

to_adjacency_matrix(store)

Return (node_ids, matrix) adjacency matrix for ML/analysis use.

Source code in src/polaroid/export.py

def to_adjacency_matrix(store: SceneStore) -> tuple[list[str], list[list[float]]]:
    """Return (node_ids, matrix) adjacency matrix for ML/analysis use."""
    nodes = store.list_nodes()
    node_ids = sorted(n.id for n in nodes)
    idx = {nid: i for i, nid in enumerate(node_ids)}
    size = len(node_ids)
    matrix: list[list[float]] = [[0.0] * size for _ in range(size)]

    for edge in store.list_edges():
        i = idx.get(edge.source_id)
        j = idx.get(edge.target_id)
        if i is not None and j is not None:
            matrix[i][j] = edge.confidence

    return node_ids, matrix

to_dot(store, graph_name='scene_graph')

Export scene graph as Graphviz DOT format. Nodes colored by node_type.

Source code in src/polaroid/export.py

def to_dot(store: SceneStore, graph_name: str = "scene_graph") -> str:
    """Export scene graph as Graphviz DOT format. Nodes colored by node_type."""
    lines: list[str] = [
        f"digraph {graph_name} {{",
        "  rankdir=LR;",
        "  node [shape=box, style=filled];",
    ]

    for node in store.list_nodes():
        color = _TYPE_COLORS.get(node.node_type, "white")
        label = f"{node.label}\\n({node.node_type})"
        lines.append(f'  "{node.id}" [label="{label}", fillcolor={color}];')

    for edge in store.list_edges():
        lines.append(f'  "{edge.source_id}" -> "{edge.target_id}" [label="{edge.relation}"];')

    lines.append("}")
    return "\n".join(lines)

to_json(store)

Export full scene graph as JSON with nodes and edges arrays.

Source code in src/polaroid/export.py

def to_json(store: SceneStore) -> str:
    """Export full scene graph as JSON with nodes and edges arrays."""
    nodes = store.list_nodes()
    edges = store.list_edges()
    data = {
        "nodes": [n.to_dict() for n in nodes],
        "edges": [e.to_dict() for e in edges],
        "node_count": len(nodes),
        "edge_count": len(edges),
    }
    return json.dumps(data, indent=2)

cluster_by_type(store)

Return {node_type: [node_ids]} grouping.

Source code in src/polaroid/stats.py

def cluster_by_type(store: SceneStore) -> dict[str, list[str]]:
    """Return {node_type: [node_ids]} grouping."""
    result: dict[str, list[str]] = {}
    for node in store.list_nodes():
        result.setdefault(node.node_type, []).append(node.id)
    return result

compute_stats(store)

Compute comprehensive statistics for the given scene store.

Source code in src/polaroid/stats.py

def compute_stats(store: SceneStore) -> GraphStats:
    """Compute comprehensive statistics for the given scene store."""
    nodes = store.list_nodes()
    edges = store.list_edges()

    node_count = store.node_count()
    edge_count = store.edge_count()

    # node_types count
    node_types: dict[str, int] = {}
    for node in nodes:
        node_types[node.node_type] = node_types.get(node.node_type, 0) + 1

    # edge_relations count
    edge_relations: dict[str, int] = {}
    for edge in edges:
        edge_relations[edge.relation] = edge_relations.get(edge.relation, 0) + 1

    # degree computation (undirected: each edge adds 1 to both endpoints)
    degree: dict[str, int] = {n.id: 0 for n in nodes}
    for edge in edges:
        if edge.source_id in degree:
            degree[edge.source_id] += 1
        if edge.target_id in degree:
            degree[edge.target_id] += 1

    total_degree = sum(degree.values())
    avg_degree = total_degree / node_count if node_count > 0 else 0.0
    max_degree = max(degree.values(), default=0)

    # Build adjacency list (undirected) for BFS
    adj: dict[str, set[str]] = {n.id: set() for n in nodes}
    for edge in edges:
        if edge.source_id in adj and edge.target_id in adj:
            adj[edge.source_id].add(edge.target_id)
            adj[edge.target_id].add(edge.source_id)

    # Connected components via BFS
    all_ids = set(n.id for n in nodes)
    visited: set[str] = set()
    components: list[set[str]] = []

    for nid in all_ids:
        if nid not in visited:
            component: set[str] = set()
            q: deque[str] = deque([nid])
            while q:
                curr = q.popleft()
                if curr in visited:
                    continue
                visited.add(curr)
                component.add(curr)
                for neighbor in adj.get(curr, set()):
                    if neighbor not in visited:
                        q.append(neighbor)
            components.append(component)

    connected_components = len(components)

    # Diameter: BFS from each node to find max shortest path
    if node_count == 0:
        diameter: int | None = None
    elif connected_components > 1:
        diameter = None
    elif node_count == 1:
        diameter = 0
    else:
        max_dist = 0
        for start in all_ids:
            dist: dict[str, int] = {start: 0}
            bfsq: deque[str] = deque([start])
            while bfsq:
                curr = bfsq.popleft()
                for neighbor in adj.get(curr, set()):
                    if neighbor not in dist:
                        dist[neighbor] = dist[curr] + 1
                        bfsq.append(neighbor)
            if dist:
                max_dist = max(max_dist, max(dist.values()))
        diameter = max_dist

    return GraphStats(
        node_count=node_count,
        edge_count=edge_count,
        node_types=node_types,
        edge_relations=edge_relations,
        avg_degree=avg_degree,
        max_degree=max_degree,
        connected_components=connected_components,
        diameter=diameter,
    )

most_connected(store, n=10)

Return top-n nodes by degree (node_id, degree).

Source code in src/polaroid/stats.py

def most_connected(store: SceneStore, n: int = 10) -> list[tuple[str, int]]:
    """Return top-n nodes by degree (node_id, degree)."""
    nodes = store.list_nodes()
    degree: dict[str, int] = {node.id: 0 for node in nodes}

    for edge in store.list_edges():
        if edge.source_id in degree:
            degree[edge.source_id] += 1
        if edge.target_id in degree:
            degree[edge.target_id] += 1

    sorted_nodes = sorted(degree.items(), key=lambda x: x[1], reverse=True)
    return sorted_nodes[:n]

extract_subgraph(store, root_id, max_depth=3)

Extract the subgraph reachable from root_id within max_depth hops.

Returns a new in-memory SceneStore.

Source code in src/polaroid/subgraph.py

def extract_subgraph(store: SceneStore, root_id: str, max_depth: int = 3) -> SceneStore:
    """Extract the subgraph reachable from root_id within max_depth hops.

    Returns a new in-memory SceneStore.
    """
    # Fetch all edges once upfront to avoid O(NxE) repeated queries inside the BFS loop.
    all_edges = store.list_edges()
    adjacency: dict[str, list[Any]] = {}
    for edge in all_edges:
        adjacency.setdefault(edge.source_id, []).append(edge)

    visited: set[str] = set()
    queue: deque[tuple[str, int]] = deque([(root_id, 0)])

    while queue:
        node_id, depth = queue.popleft()
        if node_id in visited:
            continue
        visited.add(node_id)
        if depth < max_depth:
            for edge in adjacency.get(node_id, []):
                if edge.target_id not in visited:
                    queue.append((edge.target_id, depth + 1))

    sub = SceneStore(":memory:")
    for node_id in visited:
        node = store.get_node(node_id)
        if node is not None:
            sub.upsert_node(node)

    for edge in all_edges:
        if edge.source_id in visited and edge.target_id in visited:
            sub.upsert_edge(edge)

    return sub

filter_by_type(store, node_types)

Return a new SceneStore with only nodes of the given types (and edges between them).

Source code in src/polaroid/subgraph.py

def filter_by_type(store: SceneStore, node_types: list[str]) -> SceneStore:
    """Return a new SceneStore with only nodes of the given types (and edges between them)."""
    type_set = set(node_types)
    nodes = [n for n in store.list_nodes() if n.node_type in type_set]
    node_ids = {n.id for n in nodes}

    sub = SceneStore(":memory:")
    for node in nodes:
        sub.upsert_node(node)

    for edge in store.list_edges():
        if edge.source_id in node_ids and edge.target_id in node_ids:
            sub.upsert_edge(edge)

    return sub

neighborhood(store, node_id, radius=1)

Return all node IDs within radius hops of node_id.

Source code in src/polaroid/subgraph.py

def neighborhood(store: SceneStore, node_id: str, radius: int = 1) -> list[str]:
    """Return all node IDs within `radius` hops of node_id."""
    # Fetch all edges once upfront to avoid O(radiusxE) repeated queries.
    all_edges = store.list_edges()
    adjacency: dict[str, list[str]] = {}
    for edge in all_edges:
        adjacency.setdefault(edge.source_id, []).append(edge.target_id)

    visited: set[str] = {node_id}
    frontier: set[str] = {node_id}

    for _ in range(radius):
        next_frontier: set[str] = set()
        for nid in frontier:
            for neighbor in adjacency.get(nid, []):
                if neighbor not in visited:
                    next_frontier.add(neighbor)
        visited.update(next_frontier)
        frontier = next_frontier
        if not frontier:
            break

    visited.discard(node_id)
    return list(visited)