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
python test_erubus.py     # unit tests (in-process HTTP stub; no server needed)
erubus serve &            # optional: live smoke tests run only if already up

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")
# assert_fact defaults: confidence=1.0, source="python"; omit t → server stamps now

kg.value_at("pallet-7", "at")                  # t / known_at optional query params
# {'value': 'bay-3', 'effective_confidence': 0.87, 'server_time': ...}
# effective_confidence reflects set_decay half-lives (no separate decayed() call)

# 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)                # depth-bounded neighbourhood, not the whole reachable set
# context(root, depth=2, t=None): depth bounds nodes (server clamps to 4); no known_at

kg.set_schema("located_in", functional=True)   # schema registry
kg.get_schema()                                # {"relations": [...]}
kg.retention(86400)                            # drop closed edges older than 1d (server min 1h)
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")
# Or one-shot: kg.assert_facts([{...}, {...}]) → accepted count (or ErubusError)

Batches are all-or-nothing: if any fact is invalid (e.g. a clock-skewed timestamp or out-of-range confidence) the server rejects the whole batch, nothing lands, and the client raises ErubusError (not a partial success count), so retrying is safe.

All times are UTC epoch seconds — time.time() is already correct. Omit t and the server stamps it. Optional query params (t, known_at) are omitted from the wire when not passed — never sent as null. Writes with a timestamp far in the future are rejected; call kg.clock_skew() to check your clock against the server's.