app.py

"""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

    @spaces.GPU
    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)