Add dataset ingestion and credit exchange flow

Author: Xiaofang Wu

roboclaw/account/__init__.py

@@ -1,11 +1,12 @@
 """Account credit ledger for Evo Studio billing."""
 
-from .ledger import AccountLedger, BillingRecord, PaymentOrder, Wallet
+from .ledger import AccountLedger, BillingRecord, DatasetAccessGrant, PaymentOrder, Wallet
 from .training_billing import apply_service_fee_cents, estimate_training_hold_cents, hourly_cost_from_params
 
 __all__ = [
     "AccountLedger",
     "BillingRecord",
+    "DatasetAccessGrant",
     "PaymentOrder",
     "Wallet",
     "apply_service_fee_cents",

roboclaw/account/ledger.py

@@ -10,7 +10,16 @@
 from typing import Any, Literal
 from uuid import uuid4
 
-LedgerKind = Literal["admin_recharge", "payment_recharge", "dataset_reward", "freeze", "settle", "release"]
+LedgerKind = Literal[
+    "admin_recharge",
+    "payment_recharge",
+    "dataset_reward",
+    "dataset_access_charge",
+    "dataset_access_reward",
+    "freeze",
+    "settle",
+    "release",
+]
 PaymentOrderStatus = Literal["pending", "paid", "cancelled"]
 
 
@@ -113,6 +122,28 @@ def to_dict(self) -> dict[str, Any]:
         }
 
 
+@dataclass(frozen=True)
+class DatasetAccessGrant:
+    grant_id: str
+    username: str
+    dataset_id: str
+    points_spent: int
+    contributor_username: str = ""
+    contributor_points: int = 0
+    created_at: str = ""
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "grantId": self.grant_id,
+            "username": self.username,
+            "datasetId": self.dataset_id,
+            "pointsSpent": self.points_spent,
+            "contributorUsername": self.contributor_username,
+            "contributorPoints": self.contributor_points,
+            "createdAt": self.created_at,
+        }
+
+
 class AccountLedger:
     """File-backed wallet ledger.
 
@@ -285,6 +316,87 @@ def grant_dataset_reward(
             self._save(state)
             return wallet, record, True
 
+    def redeem_dataset_access(
+        self,
+        username: str,
+        dataset_id: str,
+        price_points: int,
+        *,
+        contributor_username: str = "",
+        contributor_share_bps: int = 5000,
+        reason: str = "public dataset access",
+    ) -> tuple[Wallet, DatasetAccessGrant, BillingRecord | None, BillingRecord | None, bool]:
+        if price_points < 0:
+            raise ValueError("price_points must be non-negative")
+        if contributor_share_bps < 0 or contributor_share_bps > 10_000:
+            raise ValueError("contributor_share_bps must be between 0 and 10000")
+        username = _clean_username(username)
+        dataset_id = dataset_id.strip()
+        if not dataset_id:
+            raise ValueError("dataset_id is required")
+        contributor_username = contributor_username.strip()
+        with self._lock:
+            state = self._load()
+            for payload in state.get("datasetAccessGrants", []):
+                grant = _dataset_access_grant_from_payload(payload)
+                if grant.username == username and grant.dataset_id == dataset_id:
+                    return self._wallet_from_state(state, username), grant, None, None, False
+
+            wallet = self._wallet_from_state(state, username)
+            if wallet.reward_points < price_points:
+                raise ValueError("insufficient credit points")
+
+            wallet = Wallet(
+                username=username,
+                balance_cents=wallet.balance_cents,
+                frozen_cents=wallet.frozen_cents,
+                reward_points=wallet.reward_points - price_points,
+                updated_at=_now(),
+            )
+            buyer_record = self._append_record(
+                state,
+                wallet,
+                "dataset_access_charge",
+                -price_points,
+                reason=reason,
+                job_id=dataset_id,
+            )
+            contributor_points = 0
+            contributor_record: BillingRecord | None = None
+            if contributor_username and contributor_username != username and price_points:
+                contributor_points = price_points * contributor_share_bps // 10_000
+                if contributor_points:
+                    contributor_wallet = self._wallet_from_state(state, contributor_username)
+                    contributor_wallet = Wallet(
+                        username=contributor_username,
+                        balance_cents=contributor_wallet.balance_cents,
+                        frozen_cents=contributor_wallet.frozen_cents,
+                        reward_points=contributor_wallet.reward_points + contributor_points,
+                        updated_at=_now(),
+                    )
+                    contributor_record = self._append_record(
+                        state,
+                        contributor_wallet,
+                        "dataset_access_reward",
+                        contributor_points,
+                        reason="public dataset reuse reward",
+                        job_id=dataset_id,
+                    )
+                    self._save_wallet(state, contributor_wallet)
+            grant = DatasetAccessGrant(
+                grant_id=uuid4().hex,
+                username=username,
+                dataset_id=dataset_id,
+                points_spent=price_points,
+                contributor_username=contributor_username,
+                contributor_points=contributor_points,
+                created_at=_now(),
+            )
+            state.setdefault("datasetAccessGrants", []).append(grant.to_dict())
+            self._save_wallet(state, wallet)
+            self._save(state)
+            return wallet, grant, buyer_record, contributor_record, True
+
     def admin_recharge(self, username: str, amount_cents: int, *, reason: str = "admin recharge") -> tuple[Wallet, BillingRecord]:
         if amount_cents <= 0:
             raise ValueError("amount_cents must be positive")
@@ -610,7 +722,7 @@ def _save_wallet(self, state: dict[str, Any], wallet: Wallet) -> None:
 
     def _load(self) -> dict[str, Any]:
         if not self.path.is_file():
-            return {"wallets": {}, "records": [], "paymentOrders": []}
+            return {"wallets": {}, "records": [], "paymentOrders": [], "datasetAccessGrants": []}
         return json.loads(self.path.read_text(encoding="utf-8"))
 
     def _save(self, state: dict[str, Any]) -> None:
@@ -659,5 +771,17 @@ def _order_from_payload(payload: dict[str, Any]) -> PaymentOrder:
     )
 
 
+def _dataset_access_grant_from_payload(payload: dict[str, Any]) -> DatasetAccessGrant:
+    return DatasetAccessGrant(
+        grant_id=str(payload.get("grantId") or ""),
+        username=str(payload.get("username") or ""),
+        dataset_id=str(payload.get("datasetId") or ""),
+        points_spent=int(payload.get("pointsSpent", 0) or 0),
+        contributor_username=str(payload.get("contributorUsername") or ""),
+        contributor_points=int(payload.get("contributorPoints", 0) or 0),
+        created_at=str(payload.get("createdAt") or ""),
+    )
+
+
 def _now() -> str:
     return datetime.now(tz=timezone.utc).isoformat()

roboclaw/data/curation/service.py

@@ -243,6 +243,15 @@ def _quality_reward_multiplier(score: Any) -> float:
 
 
 def _source_reward_multiplier(info: dict[str, Any]) -> float:
+    visibility = str(
+        info.get("visibility")
+        or info.get("accessLevel")
+        or info.get("access_level")
+        or "private",
+    ).strip().lower()
+    if visibility not in {"public", "shared", "open"}:
+        return 0.0
+
     source_type = str(
         info.get("contributionSource")
         or info.get("contribution_source")

roboclaw/data/datasets.py

@@ -57,6 +57,7 @@ def datasets_root_from_manifest(manifest: Any) -> Path:
 class DatasetStats:
     total_episodes: int = 0
     total_frames: int = 0
+    total_bytes: int = 0
     fps: int = 0
     robot_type: str = ""
     features: tuple[str, ...] = ()
@@ -66,6 +67,7 @@ def to_dict(self) -> dict[str, Any]:
         return {
             "total_episodes": self.total_episodes,
             "total_frames": self.total_frames,
+            "total_bytes": self.total_bytes,
             "fps": self.fps,
             "robot_type": self.robot_type,
             "features": list(self.features),
@@ -456,6 +458,7 @@ def _read_stats(self, dataset_dir: Path, info: dict[str, Any]) -> DatasetStats:
         return DatasetStats(
             total_episodes=int(info.get("total_episodes", 0) or 0),
             total_frames=int(info.get("total_frames", 0) or 0),
+            total_bytes=_directory_size_bytes(dataset_dir),
             fps=int(info.get("fps", 0) or 0),
             robot_type=str(info.get("robot_type", "")),
             features=tuple((info.get("features") or {}).keys()),
@@ -491,3 +494,15 @@ def _local_runtime_capabilities(self) -> DatasetCapabilities:
             can_push=True,
             can_curate=True,
         )
+
+
+def _directory_size_bytes(path: Path) -> int:
+    total = 0
+    for item in path.rglob("*"):
+        if not item.is_file():
+            continue
+        try:
+            total += item.stat().st_size
+        except OSError:
+            logger.debug("Failed to stat dataset file {}", item, exc_info=True)
+    return total

roboclaw/data/ingestion.py

@@ -0,0 +1,161 @@
+"""Dataset source ingestion helpers.
+
+This module materializes external dataset sources into the local
+``DatasetCatalog`` root so the existing curation and training paths can use
+them as normal local datasets.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import urlparse
+
+from roboclaw.data.datasets import DatasetCatalog, DatasetRef
+
+
+@dataclass(frozen=True)
+class DatasetIngestSpec:
+    dataset_id: str
+    source_kind: str
+    source_uri: str
+    source_auth_ref: str = ""
+    include_videos: bool = True
+    force: bool = False
+
+
+def ingest_dataset_source(catalog: DatasetCatalog, spec: DatasetIngestSpec) -> DatasetRef:
+    """Materialize *spec* into *catalog* and return the local dataset ref."""
+    source_kind = spec.source_kind.strip().lower()
+    if source_kind in {"remote_dataset", "huggingface", "hf"}:
+        repo_id = _normalize_remote_repo(spec.source_uri)
+        return catalog.pull_dataset(
+            repo_id,
+            dataset_id=spec.dataset_id,
+            token=_resolve_auth_token(spec.source_auth_ref),
+        )
+
+    if source_kind in {"mounted_path", "local_path"}:
+        return _ingest_directory(catalog, spec)
+
+    if source_kind in {"local_upload", "local_archive", "archive"}:
+        return _ingest_archive(catalog, spec)
+
+    if source_kind in {"cloud_object", "oss_object", "s3_object", "cos_object", "gcs_object"}:
+        raise NotImplementedError(
+            "cloud object ingestion requires a configured storage provider "
+            "(OSS/S3/COS/GCS) and is not enabled in this backend yet"
+        )
+
+    raise ValueError(f"Unsupported dataset source_kind: {spec.source_kind!r}")
+
+
+def _normalize_remote_repo(source_uri: str) -> str:
+    value = source_uri.strip()
+    for prefix in ("hf://", "huggingface://"):
+        if value.startswith(prefix):
+            value = value[len(prefix):]
+            break
+    if value.startswith("https://huggingface.co/"):
+        value = value[len("https://huggingface.co/"):]
+    value = value.strip("/")
+    if value.startswith("datasets/"):
+        value = value[len("datasets/"):]
+    if not value or "/" not in value:
+        raise ValueError("remote dataset source_uri must be a HuggingFace repo id or hf://owner/name URI")
+    return value
+
+
+def _resolve_auth_token(source_auth_ref: str) -> str:
+    ref = source_auth_ref.strip()
+    if not ref or ref == "public":
+        return ""
+    env_key = "ROBOCLAW_DATASET_AUTH_" + re.sub(r"[^A-Za-z0-9]+", "_", ref).upper() + "_TOKEN"
+    token = os.environ.get(env_key, "").strip()
+    if not token:
+        raise ValueError(f"Dataset auth ref {ref!r} is not configured; expected env {env_key}")
+    return token
+
+
+def _ingest_directory(catalog: DatasetCatalog, spec: DatasetIngestSpec) -> DatasetRef:
+    source = _resolve_allowed_source_path(catalog, spec.source_uri)
+    if not source.is_dir():
+        raise ValueError(f"mounted_path source_uri must be a directory: {source}")
+    target = catalog.resolve_local_path(spec.dataset_id)
+    _prepare_target(target, spec.force)
+    shutil.copytree(source, target, dirs_exist_ok=True)
+    return catalog.require_local_dataset(spec.dataset_id)
+
+
+def _ingest_archive(catalog: DatasetCatalog, spec: DatasetIngestSpec) -> DatasetRef:
+    source = _resolve_allowed_source_path(catalog, spec.source_uri)
+    if not source.is_file():
+        raise ValueError(f"archive source_uri must be a file: {source}")
+    target = catalog.resolve_local_path(spec.dataset_id)
+    _prepare_target(target, spec.force)
+    target.mkdir(parents=True, exist_ok=True)
+    shutil.unpack_archive(str(source), str(target))
+
+    nested = _find_single_nested_dataset(target)
+    if nested is not None and nested != target:
+        temp = target.with_name(target.name + ".__ingest_tmp__")
+        if temp.exists():
+            shutil.rmtree(temp)
+        nested.rename(temp)
+        shutil.rmtree(target)
+        temp.rename(target)
+    return catalog.require_local_dataset(spec.dataset_id)
+
+
+def _prepare_target(target: Path, force: bool) -> None:
+    if target.exists():
+        if not force:
+            raise ValueError(f"Dataset target already exists: {target}")
+        shutil.rmtree(target)
+    target.parent.mkdir(parents=True, exist_ok=True)
+
+
+def _find_single_nested_dataset(target: Path) -> Path | None:
+    if (target / "meta" / "info.json").is_file():
+        return target
+    children = [child for child in target.iterdir() if child.is_dir()]
+    if len(children) != 1:
+        return None
+    child = children[0]
+    if (child / "meta" / "info.json").is_file():
+        return child
+    return None
+
+
+def _resolve_allowed_source_path(catalog: DatasetCatalog, source_uri: str) -> Path:
+    parsed = urlparse(source_uri.strip())
+    raw_path = parsed.path if parsed.scheme == "file" else source_uri
+    source = Path(raw_path).expanduser().resolve()
+    allowed_roots = _allowed_ingest_roots(catalog)
+    if not any(_is_relative_to(source, root) for root in allowed_roots):
+        roots = ", ".join(str(root) for root in allowed_roots)
+        raise ValueError(
+            f"Dataset source path is outside allowed ingest roots: {source}. "
+            f"Configure ROBOCLAW_DATASET_INGEST_ROOTS; current roots: {roots}"
+        )
+    return source
+
+
+def _allowed_ingest_roots(catalog: DatasetCatalog) -> tuple[Path, ...]:
+    configured = os.environ.get("ROBOCLAW_DATASET_INGEST_ROOTS", "").strip()
+    roots: list[Path] = []
+    if configured:
+        roots.extend(Path(value).expanduser().resolve() for value in configured.split(os.pathsep) if value.strip())
+    roots.append(catalog.root.resolve())
+    return tuple(dict.fromkeys(roots))
+
+
+def _is_relative_to(path: Path, root: Path) -> bool:
+    try:
+        path.relative_to(root)
+    except ValueError:
+        return False
+    return True