| """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", []) |
| |
| 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") |
| |
| 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)), |
| } |
|
|
