erubus 0.2.0

Client for Erubus — temporal knowledge graph for AI agents and robots

erubus (Python client)

Stdlib-only client for Erubus, a temporal knowledge graph for AI agents and robots. No dependencies.

pip install -e .          # or just copy erubus.py next to your code
erubus serve &            # the Rust binary, anywhere on your PATH
python test_erubus.py     # smoke test against it

from erubus import Erubus

kg = Erubus()                                  # http://127.0.0.1:8080
# Auth / namespace (optional; when set, sent on every request):
# kg = Erubus(token="abc123", ns="prod")       # Authorization: Bearer …, X-Erubus-Ns: …

kg.set_decay("vision:", half_life_secs=5)      # sensor beliefs go stale
kg.assert_fact("pallet-7", "at", "bay-3", confidence=0.9, source="vision:cam1")

kg.value_at("pallet-7", "at")
# {'value': 'bay-3', 'effective_confidence': 0.87, 'server_time': ...}

# What did the robot believe when it decided, not what was true:
kg.value_at("pallet-7", "at", t=20.0, known_at=decision_wall_clock)

kg.merge("bay 3", "Bay-03")                    # same entity, two spellings
kg.suggest_merge(q="bay 3")                    # fuzzy candidates (does not merge)
kg.context("pallet-7", depth=2)                # everything reachable, as of now

kg.set_schema("located_in", functional=True)   # schema registry
kg.get_schema()                                # {"relations": [...]}
kg.retention(86400)                            # drop closed edges older than 1d
kg.backup("/tmp/erubus.db")                    # consistent snapshot → local file

for fact in kg.watch("pallet-7", reconnect=True):   # live SSE stream
    print(fact)

High-rate sensors should batch — one HTTP round-trip per fact is what makes a 30Hz loop expensive:

with kg.batch(max_size=256) as b:              # flushes at max_size and on exit
    for det in detections:
        b.add(det.id, "at", det.bay, confidence=det.score, source="vision:cam1")

Batches are all-or-nothing: if any fact is invalid (e.g. a clock-skewed timestamp) the server rejects the whole batch and nothing lands, so retrying is safe.

All times are UTC epoch seconds — time.time() is already correct. Omit t and the server stamps it. Writes with a timestamp far in the future are rejected; call kg.clock_skew() to check your clock against the server's.