store

store

SQLite-backed telemetry storage.

Classes

TelemetryStore

TelemetryStore(db_path: str | Path)

Append-only SQLite store for inference telemetry records.

Source code in src/openjarvis/telemetry/store.py

def __init__(self, db_path: str | Path) -> None:
    self._db_path = str(db_path)
    self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
    self._lock = threading.Lock()
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute("PRAGMA synchronous=NORMAL")
    self._conn.execute("PRAGMA busy_timeout=5000")
    self._conn.execute(_CREATE_TABLE)
    self._conn.execute(_CREATE_MINING_STATS_TABLE)
    self._conn.commit()
    self._migrate_schema()

Functions

record

record(rec: TelemetryRecord) -> None

Persist a single telemetry record.

Source code in src/openjarvis/telemetry/store.py

def record(self, rec: TelemetryRecord) -> None:
    """Persist a single telemetry record."""
    with self._lock:
        self._conn.execute(
            _INSERT,
            (
                rec.timestamp,
                rec.model_id,
                rec.engine,
                rec.agent,
                rec.prompt_tokens,
                rec.prompt_tokens_evaluated,
                rec.completion_tokens,
                rec.total_tokens,
                rec.latency_seconds,
                rec.ttft,
                rec.cost_usd,
                rec.energy_joules,
                rec.power_watts,
                rec.gpu_utilization_pct,
                rec.gpu_memory_used_gb,
                rec.gpu_temperature_c,
                rec.throughput_tok_per_sec,
                rec.prefill_latency_seconds,
                rec.decode_latency_seconds,
                rec.energy_method,
                rec.energy_vendor,
                rec.batch_id,
                1 if rec.is_warmup else 0,
                rec.cpu_energy_joules,
                rec.gpu_energy_joules,
                rec.dram_energy_joules,
                rec.tokens_per_joule,
                rec.energy_per_output_token_joules,
                rec.throughput_per_watt,
                rec.prefill_energy_joules,
                rec.decode_energy_joules,
                rec.mean_itl_ms,
                rec.median_itl_ms,
                rec.p90_itl_ms,
                rec.p95_itl_ms,
                rec.p99_itl_ms,
                rec.std_itl_ms,
                1 if rec.is_streaming else 0,
                rec.token_counting_version,
                rec.mining_session_id,
                json.dumps(rec.metadata),
            ),
        )
        self._conn.commit()

record_mining_stats

record_mining_stats(stats: Any) -> None

Persist one mining stats snapshot.

stats is duck-typed to keep telemetry usable without importing the optional mining package at module import time.

Source code in src/openjarvis/telemetry/store.py

    def record_mining_stats(self, stats: Any) -> None:
        """Persist one mining stats snapshot.

        ``stats`` is duck-typed to keep telemetry usable without importing the
        optional mining package at module import time.
        """
        with self._lock:
            self._conn.execute(
                """\
INSERT INTO mining_stats (
    recorded_at, provider_id, shares_submitted, shares_accepted, blocks_found,
    hashrate, uptime_seconds, last_share_at, last_error, payout_target, fees_owed
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
                (
                    time.time(),
                    stats.provider_id,
                    stats.shares_submitted,
                    stats.shares_accepted,
                    stats.blocks_found,
                    stats.hashrate,
                    stats.uptime_seconds,
                    stats.last_share_at,
                    stats.last_error,
                    stats.payout_target,
                    stats.fees_owed,
                ),
            )
            self._conn.commit()

list_recent

list_recent(limit: int = 50) -> list[dict[str, Any]]

Return recent telemetry rows as dictionaries.

Source code in src/openjarvis/telemetry/store.py

def list_recent(self, limit: int = 50) -> list[dict[str, Any]]:
    """Return recent telemetry rows as dictionaries."""
    return self._select_dicts(
        "SELECT * FROM telemetry ORDER BY timestamp DESC LIMIT ?",
        (limit,),
    )

list_recent_mining_stats

list_recent_mining_stats(limit: int = 50) -> list[dict[str, Any]]

Return recent mining stats snapshots as dictionaries.

Source code in src/openjarvis/telemetry/store.py

def list_recent_mining_stats(self, limit: int = 50) -> list[dict[str, Any]]:
    """Return recent mining stats snapshots as dictionaries."""
    return self._select_dicts(
        "SELECT * FROM mining_stats ORDER BY recorded_at DESC LIMIT ?",
        (limit,),
    )

subscribe_to_bus

subscribe_to_bus(bus: EventBus) -> None

Subscribe to TELEMETRY_RECORD events on bus.

Source code in src/openjarvis/telemetry/store.py

def subscribe_to_bus(self, bus: EventBus) -> None:
    """Subscribe to ``TELEMETRY_RECORD`` events on *bus*."""
    bus.subscribe(EventType.TELEMETRY_RECORD, self._on_event)

close

close() -> None

Close the underlying SQLite connection.

Source code in src/openjarvis/telemetry/store.py

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