usage.py

"""Best-effort anonymous activity counters. Never send dataset input or content."""
import os
import queue
import threading

import requests

ORIGIN = 'https://kalanosdev-lerobot-grader.hf.space'
ENDPOINT = os.environ.get('KALANOS_API', 'https://hub.kalanos.ai').rstrip('/') + '/api/events/hf-space'
ENABLED = os.environ.get('KALANOS_USAGE_ENABLED', os.environ.get('SPACE_ID', '') == 'kalanosdev/lerobot-grader') in (True, '1', 'true')
EVENTS = {'hf_audit_started', 'hf_audit_completed', 'hf_audit_failed'}
_pending = queue.Queue(maxsize=64)
_lock = threading.Lock()
_worker = None


def _send(event):
    try:
        requests.post(ENDPOINT, json={'event': event}, headers={'Origin': ORIGIN}, timeout=2)
    except Exception:
        pass  # Measurements must never break or delay an audit.


def _drain():
    while True:
        event = _pending.get()
        try:
            _send(event)
        finally:
            _pending.task_done()


def track(event):
    global _worker
    if not ENABLED or event not in EVENTS:
        return
    try:
        with _lock:
            if _worker is None:
                _worker = threading.Thread(target=_drain, daemon=True, name='anonymous-usage')
                _worker.start()
        _pending.put_nowait(event)
    except Exception:
        pass  # Bounded queue; shed analytics when the endpoint is unavailable.