File size: 3,662 Bytes
46a15a1 1564378 46a15a1 764e398 46a15a1 1564378 b00f3e3 1564378 3d5c093 f009095 3d5c093 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """Usage-derived cost reporting and links to materialized annotations."""
import json
import math
from datetime import UTC, datetime
from studio import COPY
def usage_costs(job, report):
usage = report.get("usage") or {}
manifest = report.get("manifest") or {}
if not manifest:
try:
manifest = json.loads((getattr(job, "environment", None) or {}).get("ANNOTATION_MANIFEST", "{}"))
except (TypeError, ValueError):
manifest = {}
rate = manifest.get("costs", {}).get("cpu_rate")
seconds = getattr(getattr(job, "durations", None), "running_secs", None)
if seconds is None and getattr(job, "started_at", None):
end = getattr(job, "finished_at", None) or datetime.now(UTC)
seconds = max(0, (end - job.started_at).total_seconds())
cpu = None
if isinstance(seconds, (int, float)) and math.isfinite(seconds) and rate is not None:
cpu = math.ceil(max(0, seconds) / 60) * rate
inference = usage.get("measured_inference_usd")
return {
"inference_usd": inference,
"cpu_usd": cpu,
"total_usd": inference + cpu if inference is not None and cpu is not None else None,
"uncertain_inference_usd": max(
0, usage.get("reserved_including_uncertain_usd", 0) - (inference or 0)
),
"running_seconds": seconds,
"basis": "Reported tokens × published rates, plus HF job runtime rounded to minutes. Final HF billing may differ.",
"billing_url": "https://huggingface.co/settings/billing",
}
def _visualizer_episode(report):
manifest = report.get("manifest", {})
written = report.get("written_episode_ids", [])
# Old completed copy reports predate the explicit materialization checkpoint.
if (
not written
and report.get("status") == "completed"
and not (report.get("copy_scope") or manifest.get("copy_scope"))
):
written = [e["episode_index"] for e in report.get("episodes", []) if e["status"] == "completed"]
if manifest.get("mode") != COPY or not written:
return None
return f"{manifest['output']}/{min(written)}"
def annotation_url(report):
"""Link to the first playable exported episode's language annotations."""
episode = _visualizer_episode(report)
return f"https://lerobot-visualize-dataset.hf.space/{episode}?tab=annotations" if episode else None
def hand_visualizer_url(report):
"""Link native hand exports to the configured synchronized 3D viewer."""
if not report.get("manifest", {}).get("hand_actions"):
return None
episode = _visualizer_episode(report)
base = "https://lerobot-visualize-dataset.hf.space"
return f"{base}/{episode}?tab=hands" if episode else None
def episode_progress(report):
"""Count successful labeling separately from playable dataset exports."""
episodes = report.get("episodes", [])
expected = report["expected_episodes"]
completed = sum(e["status"] == "completed" for e in episodes)
written = report.get("written_episode_ids")
# Compatibility with reports before per-episode export checkpoints.
exported = (
len(set(written))
if written is not None
else (completed if report.get("status") == "completed" else 0)
)
return {
"expected": expected,
"completed": completed,
"exported": exported,
"export_pending": max(0, completed - exported),
"failed": sum(e["status"] == "failed" for e in episodes),
"stopped": sum(e["status"] in {"time_stopped", "budget_stopped"} for e in episodes),
"pending": max(0, expected - len(episodes)),
}
|