Spaces:
Running
Running
| """Kalanos HF Space: bounded numeric audits and published assessment examples.""" | |
| import json | |
| import logging | |
| import tempfile | |
| import threading | |
| import time | |
| import os | |
| from concurrent.futures import ThreadPoolExecutor | |
| from html import escape | |
| from urllib.parse import quote | |
| from pathlib import Path | |
| import gradio as gr | |
| import requests | |
| from cards import CSS, assistant_block, card_payload, grader_intro, grader_limits, header, kpis, render, section, share_strip | |
| from quick_grade import parse_ref, quick_grade | |
| from usage import track | |
| # ZeroGPU compatibility shim. If the Space is ever scheduled on ZeroGPU hardware, | |
| # its startup check kills the container unless at least one @spaces.GPU function | |
| # exists ("No @spaces.GPU function detected during startup"). This app is CPU-only | |
| # (the audit runs in a subprocess); the no-op below only satisfies that check. | |
| # `spaces` is absent locally, so the import is guarded. | |
| try: | |
| import spaces | |
| def _zerogpu_warmup(): | |
| return 'ok' | |
| except Exception: | |
| pass | |
| KALANOS_API = os.environ.get('KALANOS_API', 'https://hub.kalanos.ai').rstrip('/') | |
| # Featured reports: chosen for task/embodiment variety, not score. | |
| FEATURED = ('lerobot/pusht', 'lerobot/aloha_sim_insertion_human', 'RoboNex-23/Pick_orange_cube') | |
| # Example inputs that fit quick-mode bounds at the time of writing. | |
| QUICK_EXAMPLES = ('lerobot/pusht', 'lerobot/dlr_sara_pour', 'RoboCOIN/Airbot_MMK2_close_door_right') | |
| def published(ref): | |
| try: | |
| response = requests.get(f'{KALANOS_API}/d/hf/{quote(ref, safe="/")}/hf-card.json', timeout=12) | |
| if response.status_code == 200: | |
| return response.json() | |
| except (requests.RequestException, ValueError): | |
| pass | |
| return None | |
| def examples(): | |
| with ThreadPoolExecutor(max_workers=3) as pool: | |
| payloads = [p for p in pool.map(published, FEATURED) if p] | |
| head = section('Featured dataset reports') | |
| if not payloads: | |
| return head + '<div class="k-html"><p class="k-small k-muted">Featured reports are unavailable right now. You can still grade a public dataset above.</p></div>' | |
| return head + '<div class="k-html">' + kpis(payloads) + '<div class="k-cards">' + ''.join(render(p) for p in payloads) + '</div></div>' | |
| def _one(html): | |
| return f'<div class="k-html"><div class="k-cards" style="grid-template-columns:minmax(0,1fr);max-width:420px">{html}</div></div>' if html else '' | |
| def _ok(msg): | |
| return f'<span class="k-ok">{escape(msg)}</span>' | |
| def _err(msg): | |
| return f'<span class="k-error" role="alert">{escape(msg)}</span>' | |
| def _result(html, markdown, status, state): | |
| """Outputs for [out, card_text, card_acc, status, report_state]; the Markdown panel only appears with a result.""" | |
| return html, markdown, gr.update(visible=bool(markdown)), status, state | |
| def dataset_exists(ref): | |
| """True/False from the public HF API; None if HF could not be reached.""" | |
| try: | |
| r = requests.head(f'https://huggingface.co/api/datasets/{quote(ref, safe="/")}', timeout=10, allow_redirects=True) | |
| except requests.RequestException: | |
| return None | |
| if r.status_code == 200: | |
| return True | |
| if r.status_code in (401, 403, 404): | |
| return False | |
| return None | |
| NOT_FOUND = '{ref} was not found on Hugging Face, or it is private/gated. Check the org/name and try a public dataset.' | |
| # Quick audits are deterministic for (dataset, commit): reuse this Space's own fresh result for a while instead of | |
| # re-downloading and re-grading. Hub-published reports never feed this cache; only audits run here do. | |
| AUDIT_CACHE_TTL = 6 * 3600 | |
| AUDIT_CACHE_MAX = 64 | |
| _audit_cache = {} | |
| _audit_lock = threading.Lock() | |
| def current_sha(ref): | |
| """Current commit of a public dataset from the HF API without ambient credentials; None if unavailable.""" | |
| try: | |
| r = requests.get(f'https://huggingface.co/api/datasets/{quote(ref, safe="/")}', timeout=10, headers={'Authorization': ''}) | |
| return r.json().get('sha') if r.status_code == 200 else None | |
| except (requests.RequestException, ValueError): | |
| return None | |
| def cached_audit(ref, sha): | |
| if not sha: | |
| return None | |
| with _audit_lock: | |
| hit = _audit_cache.get((ref, sha)) | |
| if hit and time.monotonic() - hit[0] < AUDIT_CACHE_TTL: | |
| return hit[1] | |
| _audit_cache.pop((ref, sha), None) | |
| return None | |
| def remember_audit(ref, sha, card): | |
| if not sha: | |
| return | |
| with _audit_lock: | |
| if len(_audit_cache) >= AUDIT_CACHE_MAX: | |
| oldest = min(_audit_cache, key=lambda k: _audit_cache[k][0]) | |
| _audit_cache.pop(oldest, None) | |
| _audit_cache[(ref, sha)] = (time.monotonic(), card) | |
| def run(ref): | |
| try: | |
| ref = parse_ref(ref) | |
| except ValueError as exc: | |
| return _result('', '', _err(str(exc)), None) | |
| track('hf_audit_started') | |
| try: | |
| # Always run fresh against the current commit. Existing Hub-published reports never short-circuit grading; | |
| # only this Space's own audit of the identical commit is reused. | |
| sha = current_sha(ref) | |
| card = cached_audit(ref, sha) | |
| reused = card is not None | |
| if not reused: | |
| card = quick_grade(ref) | |
| remember_audit(ref, sha, card) | |
| payload = card_payload(card, KALANOS_API) | |
| note = 'Quick audit complete. Video: not assessed.' + (' (Result reused from an identical audit of this commit earlier today.)' if reused else '') | |
| result = _result(render(payload, published=False, detailed=True), payload['markdown'], _ok(note), card) | |
| track('hf_audit_completed') | |
| return result | |
| except ValueError as exc: | |
| track('hf_audit_failed') | |
| return _result('', '', _err(str(exc)), None) | |
| except Exception: | |
| track('hf_audit_failed') | |
| logging.exception('Quick audit failed') | |
| if dataset_exists(ref) is False: | |
| return _result('', '', _err(NOT_FOUND.format(ref=ref)), None) | |
| return _result('', '', _err('The audit could not complete. Check it is a public LeRobot v2/v3 dataset within the Space limits, or run the CLI.'), None) | |
| def lookup(ref): | |
| try: | |
| ref = parse_ref(ref) | |
| except ValueError as exc: | |
| return _result('', '', _err(str(exc)), None) | |
| payload = published(ref) | |
| if payload: | |
| return _result(render(payload, detailed=True), payload['markdown'], _ok('Stored assessment shown with its recorded date and scope. Run quick numeric check starts a fresh audit.'), payload) | |
| if dataset_exists(ref) is False: | |
| return _result('', '', _err(NOT_FOUND.format(ref=ref)), None) | |
| return _result('', '', _err(f'No published report for {ref} on Kalanos Hub yet. Run quick numeric check starts a fresh audit.'), None) | |
| def download_assessment(card): | |
| """Gradio serves this session's converted assessment and expires cached files.""" | |
| if card is None: | |
| return gr.update(value=None, visible=False) | |
| with tempfile.NamedTemporaryFile(mode='w', suffix='.json', prefix='kalanos-assessment-', delete=False) as file: | |
| json.dump(card, file, indent=2, ensure_ascii=False) | |
| download.temp_files.add(file.name) | |
| return gr.update(value=file.name, visible=True) | |
| # Publishing from the Space is intentionally disabled. The publisher upserts | |
| # reports by slug alone (hf/<org>/<name>), so a visitor's quick audit submitted | |
| # with a personal CLI key would replace the registry's full report for that | |
| # dataset. Re-enable once /api/reports scopes slug conflicts by tenant. | |
| # Until then: copy the HF card Markdown, or publish from the CLI. | |
| with gr.Blocks(title='Kalanos · Robotics Dataset Quality', delete_cache=(600, 600)) as demo: | |
| gr.HTML(header(KALANOS_API)) | |
| with gr.Column(elem_classes='k-grader'): | |
| gr.HTML(grader_intro()) | |
| with gr.Row(): | |
| ref_in = gr.Textbox(label='Public HF dataset', placeholder='lerobot/pusht', scale=4, show_label=True, container=True) | |
| go = gr.Button('Run quick numeric check', variant='primary', scale=1, elem_classes='k-primary') | |
| lookup_btn = gr.Button('Look up stored report', scale=1, elem_classes='k-secondary') | |
| gr.HTML('<p class="k-example-hint">Try an example</p>') | |
| with gr.Row(elem_classes='k-example-row'): | |
| for example in QUICK_EXAMPLES: | |
| example_button = gr.Button(example, size='sm', scale=0, elem_classes='k-example') | |
| # Fills the input client-side; never submits. | |
| example_button.click(fn=None, inputs=None, outputs=ref_in, queue=False, | |
| js=f'() => {json.dumps(example)}') | |
| gr.HTML(grader_limits()) | |
| status = gr.HTML('', elem_classes='k-status') | |
| report_state = gr.State(None, time_to_live=600) | |
| download = gr.DownloadButton('Download assessment JSON (compact)', visible=False, elem_classes='k-download') | |
| report_state.change(download_assessment, inputs=report_state, outputs=download) | |
| out = gr.HTML() | |
| with gr.Accordion('HF card Markdown', open=False, elem_classes='k-accordion', visible=False) as card_acc: | |
| copy_card = gr.Button('Copy HF dataset card', size='sm', render=False) | |
| card_text = gr.Code(language='markdown', label='Copy into your existing dataset README', interactive=False, buttons=[copy_card], elem_id='k-card-markdown') | |
| copy_card.click(fn=None, inputs=card_text, outputs=None, queue=False, | |
| js="async (markdown) => { await window.kalanosUsage?.copy(markdown, document.querySelector('#k-card-markdown .custom-button')); }") | |
| card_acc.expand(fn=None, queue=False, js="() => { window.kalanosUsage?.emit('hf_card_previewed'); }") | |
| featured = gr.HTML(section('Featured dataset reports')) | |
| demo.load(examples, outputs=featured) | |
| with gr.Accordion('Use with your AI assistant (Claude, ChatGPT) · read-only', open=False, elem_classes='k-accordion', elem_id='k-assistant'): | |
| gr.HTML(assistant_block()) | |
| outputs = [out, card_text, card_acc, status, report_state] | |
| go.click(run, inputs=ref_in, outputs=outputs, concurrency_limit=1, concurrency_id='audit') | |
| ref_in.submit(run, inputs=ref_in, outputs=outputs, concurrency_limit=1, concurrency_id='audit') | |
| lookup_btn.click(lookup, inputs=ref_in, outputs=outputs) | |
| gr.HTML(share_strip(KALANOS_API)) | |
| gr.HTML('<p class="k-small k-muted">Anonymous activity counts help improve this demo. Dataset names and card contents are not included.</p>') | |
| THEME = gr.themes.Base( | |
| primary_hue=gr.themes.colors.green, neutral_hue=gr.themes.colors.stone, | |
| font=[gr.themes.GoogleFont('IBM Plex Sans'), 'sans-serif'], | |
| font_mono=[gr.themes.GoogleFont('IBM Plex Mono'), 'monospace'], | |
| ).set(body_background_fill='#f6f7f2', body_text_color='#182f29', block_radius='8px', button_large_radius='6px', | |
| input_radius='3px', input_background_fill='#ffffff', input_border_color='#d2d3cb') | |
| # Light-only: Gradio adds `.dark` to <body> from the viewer's system setting (or | |
| # ?__theme=dark); every surface here is pinned light, so dark text variables would | |
| # produce light-on-light text. Remove the class at load and keep it off. | |
| FORCE_LIGHT_JS = '''() => { | |
| const off = () => { document.body.classList.remove('dark'); document.documentElement.classList.remove('dark'); }; | |
| off(); new MutationObserver(off).observe(document.body, { attributes: true, attributeFilter: ['class'] }); | |
| }''' | |
| FORCE_LIGHT_HEAD = '<meta name="color-scheme" content="light">' | |
| APP_JS = FORCE_LIGHT_JS.replace('() => {', '() => {\n' + Path(__file__).with_name('usage.js').read_text() + '\n', 1) | |
| if __name__ == '__main__': | |
| demo.queue(max_size=4).launch(css=CSS, theme=THEME, js=APP_JS, head=FORCE_LIGHT_HEAD) | |