Spaces:
Running
Running
Numeric quick audits, featured report KPIs from hub hf-card.json, HF card export (publish disabled pending tenant-scoped slugs)
dc7468e | """Bounded public LeRobot audit. No video files or ambient HF credentials.""" | |
| import json | |
| import os | |
| from pathlib import Path, PurePosixPath | |
| import re | |
| import subprocess | |
| import sys | |
| import tempfile | |
| MAX_BYTES = 128 * 1024 * 1024 | |
| MAX_FILES = 256 | |
| MAX_EPISODES = 500 | |
| TIMEOUT = 180 | |
| REF = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*\Z") | |
| def parse_ref(raw): | |
| ref = (raw or '').strip().removeprefix('https://huggingface.co/datasets/').rstrip('/') | |
| if not REF.fullmatch(ref) or '..' in ref: | |
| raise ValueError('Enter a public dataset as org/name or its HF dataset URL.') | |
| return ref | |
| def allowed_file(name): | |
| path = PurePosixPath(name) | |
| return (not path.is_absolute() and '..' not in path.parts and | |
| ((name.startswith('data/') and name.endswith('.parquet')) or | |
| (name.startswith('meta/') and path.suffix in {'.json', '.jsonl', '.parquet'}))) | |
| def check_metadata(meta): | |
| if not str(meta.get('codebase_version', '')).startswith(('v2', 'v3')): | |
| raise ValueError('Quick mode supports LeRobot v2/v3. Use the CLI for this format.') | |
| if not 0 < meta.get('total_episodes', 0) <= MAX_EPISODES: | |
| raise ValueError(f'Quick mode accepts up to {MAX_EPISODES} episodes. Use the CLI for this dataset.') | |
| if any(f.get('dtype') == 'image' for f in meta.get('features', {}).values()): | |
| raise ValueError('Embedded images require a CLI audit; quick mode excludes media.') | |
| for key in ('data_path', 'video_path'): | |
| path = str(meta.get(key) or '') | |
| if '://' in path or PurePosixPath(path).is_absolute() or '..' in PurePosixPath(path).parts: | |
| raise ValueError('Dataset contains unsupported file paths. Use the CLI.') | |
| def grade_local(root, ref, revision, license_name, byte_count, file_count): | |
| from kalanos import grade, SourceInfo | |
| from publisher_grader.convert import convert | |
| from publisher_grader.provenance import Provenance | |
| # All selected numeric shards are assessed; never silently sample episodes. | |
| report = grade(root, bundle='numeric-core-v1', | |
| sidecar=False, full_frame_scan=False) | |
| report.source = SourceInfo(uri=f'hf://datasets/{ref}@{revision}', protocol='hf', | |
| repo_id=ref, revision=revision, | |
| size_bytes=byte_count, file_count=file_count) | |
| report.run.source.revision = revision | |
| card = convert(report, Provenance(kind='hf', ref=ref, | |
| url=f'https://huggingface.co/datasets/{ref}', | |
| license=license_name or 'unknown', gated=False)).model_dump(mode='json') | |
| card['report_id'] = report.run.id | |
| card['notes'].append('HF Space quick audit: all downloaded numeric shards assessed; ' | |
| 'video and image files excluded. Video: not assessed. ' | |
| 'Storage size describes downloaded numeric/metadata files, not the full HF repository.') | |
| return card | |
| def worker(ref, destination): | |
| import requests | |
| from huggingface_hub import HfApi, hf_hub_url | |
| api = HfApi(token=False) | |
| info = api.dataset_info(ref, token=False) | |
| if info.private or info.gated: | |
| raise ValueError('Quick mode supports public, ungated datasets. Use the CLI for private or gated data.') | |
| revision = info.sha | |
| if not revision: | |
| raise ValueError('Could not resolve a dataset revision.') | |
| selected = [] | |
| # Cap listing work too; a huge media tree should not exhaust Space memory. | |
| for index, entry in enumerate(api.list_repo_tree(ref, repo_type='dataset', revision=revision, recursive=True, token=False)): | |
| if index >= 10000: | |
| raise ValueError('Dataset tree exceeds the quick-mode limit. Use the CLI.') | |
| if allowed_file(entry.path) and hasattr(entry, 'size'): | |
| selected.append(entry) | |
| if len(selected) > MAX_FILES or sum(e.size for e in selected) > MAX_BYTES: | |
| raise ValueError('Numeric data exceeds 128 MiB / 256 files. Use the CLI.') | |
| if not any(e.path == 'meta/info.json' for e in selected): | |
| raise ValueError('No LeRobot meta/info.json found.') | |
| with tempfile.TemporaryDirectory(prefix='kalanos-numeric-') as tmp: | |
| root = Path(tmp) | |
| downloaded = 0 | |
| # Validate the manifest before downloading shards. | |
| selected.sort(key=lambda e: e.path != 'meta/info.json') | |
| for entry in selected: | |
| target = root / entry.path | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| with requests.get(hf_hub_url(ref, entry.path, repo_type='dataset', revision=revision), | |
| stream=True, timeout=(10, 25)) as response: | |
| response.raise_for_status() | |
| with target.open('wb') as output: | |
| for chunk in response.iter_content(1024 * 1024): | |
| downloaded += len(chunk) | |
| if downloaded > MAX_BYTES: | |
| raise ValueError('Download exceeds the quick-mode limit. Use the CLI.') | |
| output.write(chunk) | |
| if entry.path == 'meta/info.json': | |
| check_metadata(json.loads(target.read_text())) | |
| card_data = info.card_data or {} | |
| card = grade_local(root, ref, revision, card_data.get('license', 'unknown'), downloaded, len(selected)) | |
| Path(destination).write_text(json.dumps(card)) | |
| def quick_grade(ref): | |
| ref = parse_ref(ref) | |
| with tempfile.TemporaryDirectory(prefix='kalanos-job-') as tmp: | |
| output = Path(tmp) / 'card.json' | |
| env = {k: v for k, v in os.environ.items() if not k.startswith(('HF_', 'HUGGING_FACE_', 'KALANOS_')) | |
| and k != 'INGEST_API_KEY'} | |
| env.update(HF_HUB_DISABLE_IMPLICIT_TOKEN='1', HF_HOME=tmp, | |
| KALANOS_PLUGIN_DIR=str(Path(tmp) / 'plugins'), OMP_NUM_THREADS='2') | |
| try: | |
| result = subprocess.run([sys.executable, str(Path(__file__).resolve()), ref, str(output)], | |
| env=env, capture_output=True, text=True, timeout=TIMEOUT) | |
| except subprocess.TimeoutExpired: | |
| raise ValueError('Quick audit exceeded 3 minutes. Run this dataset with the CLI.') from None | |
| if result.returncode: | |
| # Only intentional validation errors are user-facing; never expose tracebacks or paths. | |
| error_file = Path(str(output) + '.error') | |
| raise ValueError(error_file.read_text() if error_file.exists() else | |
| 'The quick audit could not complete. Use the CLI for details.') | |
| return json.loads(output.read_text()) | |
| if __name__ == '__main__': | |
| try: | |
| worker(parse_ref(sys.argv[1]), sys.argv[2]) | |
| except ValueError as exc: | |
| Path(sys.argv[2] + '.error').write_text(str(exc)) | |
| sys.exit(1) | |