Spaces:
Running
Running
| """Kalanos HF Space: bounded numeric audits and published assessment examples.""" | |
| import json | |
| import logging | |
| import tempfile | |
| 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, card_payload, grader_intro, header, kpis, render, section, share_strip | |
| from quick_grade import parse_ref, quick_grade | |
| from usage import track | |
| CSS += '\n#k-card-markdown .custom-button-label { font-size:0; } #k-card-markdown .custom-button-label::before { content:"⧉"; font-size:18px; }' | |
| # 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 = ('lerobot/aloha_static_towel', 'lerobot/pusht', 'lerobot/aloha_sim_insertion_human') | |
| 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.' | |
| 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. Existing published reports never short-circuit grading. | |
| card = quick_grade(ref) | |
| payload = card_payload(card, KALANOS_API) | |
| result = _result(render(payload, published=False, detailed=True), payload['markdown'], _ok('Quick audit complete. Video: not assessed.'), 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. Grade dataset runs 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. Grade dataset runs a fresh quick 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=False, container=False) | |
| go = gr.Button('Grade dataset', variant='primary', scale=1, elem_classes='k-primary') | |
| # Secondary action stays clear; report cards themselves use compact icons. | |
| lookup_btn = gr.Button('View published report', scale=1, elem_classes='k-secondary') | |
| 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) | |
| 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) | |
| 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.slate, 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='#f2efe6', body_text_color='#202622', block_radius='0px', button_large_radius='3px', | |
| 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) | |