Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
Download app/envs/rows.py from FineEnvs/RL-Explorer: direct link, hf CLI and curl.
- Browser
- Download file 34.6 kB
-
https://huggingface.co/spaces/FineEnvs/RL-Explorer/resolve/main/app/envs/rows.py
- Command line
-
hf download hf://spaces/FineEnvs/RL-Explorer/app/envs/rows.py
-
curl -L -o rows.py https://huggingface.co/spaces/FineEnvs/RL-Explorer/resolve/main/app/envs/rows.py
34.6 kB
| """Environments whose tasks are dataset rows (Verifiers, NeMo Gym, verl, MiMo's raw release, Harbor tasks packed into | |
| parquet, traces, any table of prompts). Rows come from the Hub's dataset viewer, or straight from the dataset's files | |
| when the viewer can't read it (direct.py); a row reader (processors.py) turns each into a task. `RowsAdapter`, at the | |
| end, puts this behind the environment contract (contract.py): a task's ref is `<split>/<row>`, or | |
| `<config>/<split>/<row>` for a dataset with several configs. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import time | |
| from collections import Counter | |
| from typing import Any | |
| from . import contract as c | |
| from . import direct, viewer | |
| from .base import ANSWER, FACET_SKIP, Dataset, Processor, parse_json, retitle | |
| from .direct import DirectError | |
| from .processors import HIDDEN, PROCESSORS | |
| from .viewer import ViewerError | |
| __all__ = ["describe", "rows", "task", "task_file", "packed", "materialize", "ViewerError", "DirectError", "pick"] | |
| PAGE = 24 | |
| def pick(ds: Dataset) -> Processor: | |
| return max(PROCESSORS, key=lambda p: p.match(ds)) | |
| # ── where rows come from ───────────────────────────────────────────────────── | |
| class Viewer: | |
| kind = "viewer" | |
| def __init__(self, spec: str, tok: str | None): | |
| self.spec, self.tok = spec, tok | |
| def splits(self) -> list[dict[str, str]]: | |
| return viewer.splits(self.spec, self.tok) | |
| def page(self, config: str, split: str, offset: int, length: int) -> dict[str, Any]: | |
| return viewer.rows(self.spec, config, split, offset, length, self.tok) | |
| def search(self, config, split, q, offset): | |
| try: | |
| if _broken.get((self.spec, "search"), 0) > time.time(): | |
| raise ViewerError("not ready") | |
| return _visible_hits(viewer.search(self.spec, config, split, q, offset, PAGE, self.tok), q, offset), None | |
| except ViewerError: # its search index isn't ready (or is off): a small split is searched here instead | |
| _broken[(self.spec, "search")] = time.time() + 600 | |
| ql = q.lower() | |
| return self._local(config, split, offset, lambda r: ql in json.dumps(_searchable(r), ensure_ascii=False, default=str).lower()) | |
| def filter(self, config, split, filters, columns, offset): | |
| try: | |
| if _broken.get((self.spec, "filter"), 0) > time.time(): | |
| raise ViewerError("not ready") | |
| return viewer.filter_rows(self.spec, config, split, _where(filters, columns), offset, PAGE, self.tok), None | |
| except ViewerError: | |
| _broken[(self.spec, "filter")] = time.time() + 600 | |
| return self._local(config, split, offset, lambda r: _matches(r, filters)) | |
| def both(self, config, split, q, filters, columns, offset): | |
| """A search within a filter: the viewer does one or the other, so a small split is read here and matched on | |
| both; a big one is searched, with the filter left off (and a note saying so).""" | |
| ql = q.lower() | |
| try: | |
| return self._local(config, split, offset, lambda r: _matches(r, filters) and ql in json.dumps(_searchable(r), ensure_ascii=False, default=str).lower()) | |
| except ViewerError: | |
| page, note = self.search(config, split, q, offset) | |
| return page, "This dataset is too big to search within a filter here, so the filter is left off this search." | |
| def _local(self, config, split, offset, match): | |
| rows = _all_rows(self.spec, config, split, self.tok) | |
| hits = [(i, r, []) for i, r in rows if match(r)] | |
| return {"rows": hits[offset:offset + PAGE], "total": len(hits), "features": []}, None | |
| def facets(self, config, split, roles, sample) -> list[dict[str, Any]]: | |
| out = [] | |
| skip = {roles.get("task"), roles.get("id"), roles.get("title"), *roles.get("answer", [])} | |
| for st in viewer.statistics(self.spec, config, split, self.tok): | |
| col, kind, stats = st.get("column_name"), st.get("column_type"), st.get("column_statistics") or {} | |
| if col in skip or ANSWER.search(col or "") or FACET_SKIP.search(col or ""): | |
| continue | |
| freq = stats.get("frequencies") | |
| if kind in ("string_label", "bool", "class_label") and isinstance(freq, dict) and 2 <= len(freq) <= 60: | |
| out.append({"column": col, "values": sorted(([str(k), v] for k, v in freq.items()), key=lambda x: -x[1])[:40], "sampled": False}) | |
| return out[:8] or _sample_facets(sample, roles) | |
| def can_search(self) -> bool: | |
| return bool(viewer.valid(self.spec, self.tok).get("search")) | |
| class Files: | |
| kind = "files" | |
| def __init__(self, spec: str, meta: dict[str, Any], tok: str | None): | |
| self.spec, self.meta, self.tok = spec, meta, tok | |
| self.configs = direct.data_files(spec, meta, tok) | |
| if not self.configs: | |
| raise DirectError("no data files (JSON Lines, JSON, parquet or CSV) in this dataset") | |
| def splits(self) -> list[dict[str, str]]: | |
| return [{"config": c, "split": s} for c, ss in self.configs.items() for s in ss] | |
| def source(self, config: str, split: str) -> direct.Source: | |
| return direct.Source(self.spec, self.meta, self.configs.get(config, {}).get(split) or [], self.tok) | |
| def page(self, config, split, offset, length): | |
| src = self.source(config, split) | |
| got = src.read(offset, length) | |
| return {"rows": [(i, r, []) for i, r in got], "total": src.total(), "features": _features([r for _, r in got])} | |
| def _scan(self, config, split, offset, match, what): | |
| hits, cut = direct.scan(self.source(config, split), match) | |
| note = f"{what} looked through the first {direct.SCAN:,} rows only." if cut else None | |
| return {"rows": [(i, r, []) for i, r in hits[offset:offset + PAGE]], "total": len(hits), "features": []}, note | |
| def search(self, config, split, q, offset): | |
| ql = q.lower() | |
| return self._scan(config, split, offset, lambda r: ql in json.dumps(_searchable(r), ensure_ascii=False, default=str).lower(), "Search") | |
| def filter(self, config, split, filters, columns, offset): | |
| return self._scan(config, split, offset, lambda r: _matches(r, filters), "This filter") | |
| def both(self, config, split, q, filters, columns, offset): | |
| ql = q.lower() | |
| return self._scan(config, split, offset, lambda r: _matches(r, filters) and ql in json.dumps(_searchable(r), ensure_ascii=False, default=str).lower(), "Search") | |
| def facets(self, config, split, roles, sample): | |
| return _sample_facets(sample, roles) | |
| def can_search(self) -> bool: | |
| return True | |
| LOCAL_MAX = 20_000 # rows of a split worth reading whole, to search and filter here when the viewer can't | |
| _broken: dict[tuple[str, str], float] = {} | |
| _choice: dict[tuple[str, str], tuple[str, float]] = {} # (dataset, revision) -> (viewer|files, until) # (dataset, "search"|"filter") -> until when the viewer's is skipped | |
| def _word_in(w: str, text: str) -> bool: | |
| """The viewer's search stems words ("graders" finds "grader"): the same allowance here, and no looser.""" | |
| if w in text: | |
| return True | |
| for suf in ("ing", "es", "ed", "s"): | |
| if len(w) > len(suf) + 2 and w.endswith(suf) and w[: -len(suf)] in text: | |
| return True | |
| return False | |
| def _visible_hits(page: dict[str, Any], q: str, offset: int) -> dict[str, Any]: | |
| """The dataset viewer's search looks at every column, the withheld ones too: a row it found only through its answer | |
| would tell anyone (an agent over MCP, say) which rows hold that answer. Only rows whose visible fields have every | |
| word are kept, and when any was dropped the total stops at what's shown, so the count says nothing either.""" | |
| words = [w for w in q.lower().split() if w] | |
| kept = [h for h in page["rows"] if all(_word_in(w, json.dumps(_searchable(h[1]), ensure_ascii=False, default=str).lower()) for w in words)] | |
| if len(kept) == len(page["rows"]): | |
| return page | |
| return {**page, "rows": kept, "total": offset + len(kept)} | |
| def _searchable(row: dict[str, Any]) -> dict[str, Any]: | |
| """What search looks at: the row without its answers (searching for an answer mustn't find its rows).""" | |
| from .base import withhold | |
| return withhold(row, hide=HIDDEN)[0] | |
| def _all_rows(spec: str, config: str, split: str, tok: str | None) -> list[tuple[int, dict[str, Any]]]: | |
| """Every row of a small split, through the viewer's row pages (8 at a time), kept for half an hour.""" | |
| from concurrent.futures import ThreadPoolExecutor | |
| from .. import catalog | |
| def fetch(): | |
| first = viewer.rows(spec, config, split, 0, 100, tok) | |
| total = first["total"] or 0 | |
| if total > LOCAL_MAX: | |
| raise ViewerError(f"the dataset viewer can't filter this dataset right now, and its {total:,} rows are too many to read here") | |
| pages = [first] + list(ThreadPoolExecutor(8).map(lambda o: viewer.rows(spec, config, split, o, 100, tok), range(100, total, 100))) | |
| return [(i, r) for pg in pages for i, r, _ in pg["rows"]] | |
| return catalog._cached(("all-rows", spec, config, split), 1800, fetch, big=True) | |
| def _features(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| names: list[str] = [] | |
| for r in rows: | |
| names += [k for k in r if k not in names] | |
| return [{"name": n, "type": {}} for n in names] | |
| def _sample_facets(sample: list[dict[str, Any]], roles: dict[str, Any]) -> list[dict[str, Any]]: | |
| """Facets from the first rows, when there are no statistics: short labels that repeat.""" | |
| skip = {roles.get("task"), roles.get("id"), roles.get("title"), *roles.get("answer", [])} | |
| out = [] | |
| for col in _features(sample): | |
| c = col["name"] | |
| if c in skip or ANSWER.search(c) or FACET_SKIP.search(c): | |
| continue | |
| vals = [r.get(c) for r in sample if isinstance(r.get(c), (str, bool)) and len(str(r.get(c))) <= 40] | |
| cnt = Counter(str(v) for v in vals) | |
| if len(vals) >= len(sample) * 0.8 and 2 <= len(cnt) <= 20 and max(cnt.values()) >= 2: | |
| out.append({"column": c, "values": [[k, v] for k, v in cnt.most_common(20)], "sampled": True}) | |
| return out[:6] | |
| def _where(filters: dict[str, Any], columns: list[str]) -> str: | |
| """A filter for the viewer from facet picks ({column: value or [values]}): only known columns, values quoted; | |
| several values of one column are any of them, columns are all of them.""" | |
| parts = [] | |
| for col, vals in filters.items(): | |
| vals = [vals] if isinstance(vals, str) else [v for v in vals if isinstance(v, str)] if isinstance(vals, list) else [] | |
| vals = [v for v in vals if len(v) <= 300][:20] | |
| if col not in columns or not vals: | |
| continue | |
| name = '"%s"' % col.replace('"', '""') | |
| ors = " OR ".join("%s='%s'" % (name, v.replace("'", "''")) for v in vals) | |
| parts.append(ors if len(vals) == 1 else f"({ors})") | |
| return " AND ".join(parts) | |
| def _matches(row: dict[str, Any], filters: dict[str, Any]) -> bool: | |
| for k, vals in filters.items(): | |
| vals = [vals] if isinstance(vals, str) else vals | |
| if str(row.get(k)) not in vals: | |
| return False | |
| return True | |
| def _info(spec: str, token: str | None) -> tuple[dict[str, Any], str | None]: | |
| """The dataset's Hub record (checking the visitor may read it) and the token to read its rows with.""" | |
| from .. import catalog | |
| meta = catalog.info(spec, token) | |
| return meta, (token if meta.get("restricted") else None) | |
| def _backend(spec: str, token: str | None): | |
| """The dataset viewer when it can read this dataset, else its files. Checks the visitor may read it.""" | |
| from .. import catalog | |
| meta = catalog.info(spec, token) | |
| tok = token if meta.get("restricted") else None | |
| def choose(): | |
| v = Viewer(spec, tok) | |
| for attempt in range(2): # one slow answer shouldn't send a dataset to its files | |
| try: | |
| sp = v.splits() | |
| if sp: | |
| v.page(sp[0]["config"], sp[0]["split"], 0, 1) | |
| return "viewer" | |
| break | |
| except ViewerError as e: | |
| if e.status == 404: # the viewer can't read it at all | |
| break | |
| return "files" | |
| key = (spec, meta["sha"]) | |
| hit = _choice.get(key) | |
| if not hit or hit[1] < time.time(): | |
| which = choose() | |
| _choice[key] = (which, time.time() + (1800 if which == "viewer" else 120)) # files: try the viewer again soon | |
| which = _choice[key][0] | |
| return meta, (Viewer(spec, tok) if which == "viewer" else Files(spec, meta, tok)) | |
| _LEFTOVER = re.compile(r"(dropped|discarded|excluded|removed|failed|invalid|rejected|unsolved)", re.IGNORECASE) | |
| def _first(splits: list[dict[str, str]]) -> dict[str, str]: | |
| """The split a dataset opens on: its first config's train split, else its first that isn't a leftover (rows | |
| dropped from the dataset proper).""" | |
| first = [s for s in splits if s["config"] == splits[0]["config"]] | |
| return next((s for s in first if s["split"] == "train"), None) or next((s for s in first if not _LEFTOVER.search(s["split"])), None) or splits[0] | |
| _split_files: dict[tuple[str, str, str, str], float] = {} # (dataset, revision, config, split) -> until: read from files | |
| def _first_page(spec: str, meta: dict[str, Any], be, pair: dict[str, str]): | |
| """The split's first rows, and the backend that reads that split. The viewer reads a dataset config by config, and | |
| can fail on one ("The dataset generation failed": a schema it can't cast) while reading the rest: that split is | |
| then read from its files, and remembered for a while, so its pages, tasks and search come from there too.""" | |
| key = (spec, str(meta.get("sha")), pair["config"], pair["split"]) | |
| if be.kind == "viewer" and _split_files.get(key, 0) > time.time(): | |
| try: | |
| fb = Files(spec, meta, be.tok) | |
| return fb, fb.page(pair["config"], pair["split"], 0, 20) | |
| except DirectError: | |
| _split_files.pop(key, None) | |
| try: | |
| return be, be.page(pair["config"], pair["split"], 0, 20) | |
| except ViewerError as e: | |
| if be.kind != "viewer" or e.status == 404: | |
| raise | |
| try: | |
| fb = Files(spec, meta, be.tok) | |
| if not any(s["config"] == pair["config"] and s["split"] == pair["split"] for s in fb.splits()): | |
| raise e from None | |
| page = fb.page(pair["config"], pair["split"], 0, 20) | |
| except DirectError: | |
| raise e from None | |
| _split_files[key] = time.time() + 1800 | |
| return fb, page | |
| def _dataset(spec: str, config: str | None, split: str | None, token: str | None): | |
| meta, be = _backend(spec, token) | |
| splits = be.splits() | |
| if not splits: | |
| raise ViewerError("this dataset has no rows to read", 404) | |
| pair = next((s for s in splits if s["config"] == config and s["split"] == split), None) \ | |
| or next((s for s in splits if s["config"] == config), None) or _first(splits) | |
| be, page = _first_page(spec, meta, be, pair) | |
| ds = Dataset(spec=spec, tags=meta.get("all_tags") or meta.get("tags") or [], card=meta, splits=splits, features=page["features"] or _features([r for _, r, _ in page["rows"]]), | |
| sample=[r for _, r, _ in page["rows"]]) | |
| proc = pick(ds) | |
| roles = proc.roles(ds) | |
| roles["_config"], roles["_split"] = pair["config"], pair["split"] | |
| return be, ds, pair["config"], pair["split"], proc, roles, page["total"] | |
| def describe(spec: str, token: str | None = None, config: str | None = None, split: str | None = None, overview: bool = False) -> dict[str, Any]: | |
| """The dataset as read: its reader, configs, roles and facets; with `overview`, the reader's own sections for the | |
| dataset's page (only the page asks: lists and random picks don't pay for them).""" | |
| be, ds, config, split, proc, roles, total = _dataset(spec, config, split, token) | |
| configs: dict[str, list[str]] = {} | |
| for s in ds.splits: | |
| configs.setdefault(s["config"], []).append(s["split"]) | |
| sections: list[dict[str, Any]] = [] | |
| if overview: | |
| try: # from what's read already: the first rows and the viewer's statistics (a framework's grading, its run commands) | |
| stats = viewer.statistics(spec, config, split, be.tok) if be.kind == "viewer" else [] | |
| sections = proc.overview(ds, roles, stats, config, split) | |
| except Exception: # noqa: BLE001 - an overview that can't be worked out isn't shown; the rows still are | |
| sections = [] | |
| fw = proc.framework_for(ds) if hasattr(proc, "framework_for") else proc.framework | |
| return { | |
| "processor": {"id": proc.id, "name": proc.name, "framework": fw, "about": proc.about}, | |
| "configs": [{"config": c, "splits": ss} for c, ss in configs.items()], | |
| "config": config, "split": split, "total": total, "source": be.kind, | |
| "roles": {k: v for k, v in roles.items() if not k.startswith("_")}, | |
| "columns": ds.columns, | |
| "facets": be.facets(config, split, roles, ds.sample), | |
| "search": be.can_search(), | |
| "overview": sections, | |
| } | |
| def rows(spec: str, config: str, split: str, offset: int = 0, q: str = "", filters: dict[str, str] | None = None, | |
| token: str | None = None) -> dict[str, Any]: | |
| be, ds, config, split, proc, roles, total = _dataset(spec, config, split, token) | |
| filters = {k: v for k, v in (filters or {}).items() if k in ds.columns} | |
| note = None | |
| try: | |
| if q.strip() and filters: | |
| page, note = be.both(config, split, q.strip()[:200], filters, ds.columns, offset) | |
| mode = "search" | |
| elif q.strip(): | |
| page, note = be.search(config, split, q.strip()[:200], offset) | |
| mode = "search" | |
| elif filters: | |
| page, note = be.filter(config, split, filters, ds.columns, offset) | |
| mode = "filter" | |
| else: | |
| page, mode = be.page(config, split, offset, PAGE), "rows" | |
| except (ViewerError, DirectError) as e: | |
| if not (q.strip() or filters): | |
| raise | |
| page = be.page(config, split, offset, PAGE) | |
| mode, note = "rows", f"{'Search' if q.strip() else 'Filtering'} isn't available for this dataset right now ({e}); showing every row." | |
| cards = [] | |
| for i, row, _ in page["rows"]: | |
| try: | |
| cards.append(proc.card(row, i, roles)) | |
| except Exception: # noqa: BLE001 - one odd row shows as a plain row, it doesn't sink the page | |
| cards.append({"i": i, "id": None, "title": f"Row {i}", "snippet": "", "chips": []}) | |
| retitle(cards) | |
| return {"rows": cards, "total": page["total"], "offset": offset, "page": PAGE, "mode": mode, "note": note, | |
| "config": config, "split": split} | |
| def _row(spec: str, config: str, split: str, index: int, token: str | None): | |
| be, ds, config, split, proc, roles, total = _dataset(spec, config, split, token) | |
| if index < 0 or (total is not None and index >= total): | |
| raise LookupError("no such row") | |
| page = be.page(config, split, index, 1) | |
| if not page["rows"]: | |
| raise LookupError("no such row") | |
| _, row, truncated = page["rows"][0] | |
| return ds, config, split, proc, roles, total, row, truncated | |
| def task(spec: str, config: str, split: str, index: int, token: str | None = None) -> dict[str, Any]: | |
| ds, config, split, proc, roles, total, row, truncated = _row(spec, config, split, index, token) | |
| view = proc.view(row, index, ds, roles) | |
| view.update(spec=spec, config=config, split=split, index=index, total=total, truncated=truncated, | |
| processor={"id": proc.id, "name": proc.name, "framework": proc.framework, "about": proc.about}, | |
| restricted=bool(_info(spec, token)[0].get("restricted"))) | |
| view.setdefault("run", None) | |
| return view | |
| def task_file(spec: str, config: str, split: str, index: int, rel: str, token: str | None = None) -> dict[str, Any]: | |
| ds, config, split, proc, roles, total, row, _ = _row(spec, config, split, index, token) | |
| if not hasattr(proc, "file"): | |
| raise LookupError("this dataset's rows have no files") | |
| return proc.file(row, ds, re.sub(r"^/+", "", rel or "")) | |
| # ── running a Harbor task packed into a row ────────────────────────────────── | |
| def packed(spec: str, config: str, split: str, index: int, token: str | None = None) -> dict[str, Any]: | |
| """A packed Harbor task as the runner needs it: its files, title, and whether (and how) it can run here.""" | |
| import tomllib | |
| from .. import catalog | |
| from .processors import PackedHarbor, unpack | |
| ds, config, split, proc, roles, total, row, _ = _row(spec, config, split, index, token) | |
| if not isinstance(proc, PackedHarbor): | |
| raise LookupError("this dataset's rows aren't Harbor tasks") | |
| files = unpack(row.get(proc.blob(ds))) | |
| if "task.toml" not in files: | |
| raise LookupError("this row's archive holds no Harbor task") | |
| texts = {n: b.decode("utf-8", "replace") for n, b in files.items() if b"\x00" not in b[:4096]} | |
| try: | |
| doc = tomllib.loads(texts.get("task.toml", "")) | |
| except tomllib.TOMLDecodeError: | |
| raise LookupError("this task's task.toml can't be read") from None | |
| env = doc.get("environment") or {} | |
| look = {"env": {"image": env.get("docker_image"), "network": catalog._network(env), | |
| "compose": any(re.match(r"environment/(docker-)?compose\.ya?ml$", n) for n in files)}} | |
| meta = _info(spec, token)[0] | |
| view = proc.view(row, index, ds, roles) | |
| return {"title": view["title"], "files": files, "runnable": catalog.runnable(look, texts), "image": env.get("docker_image"), | |
| "restricted": bool(meta.get("restricted")), "sha": meta["sha"], "bytes": sum(len(b) for b in files.values()), | |
| "verifier_env": sorted(((doc.get("verifier") or {}).get("env") or {}).keys())} | |
| def materialize(spec: str, config: str, split: str, index: int, token: str | None = None): | |
| """The packed task written out as a folder for the runner, without its reference solution.""" | |
| import shutil | |
| from .. import catalog, config as cfg | |
| task = packed(spec, config, split, index, token) | |
| root = cfg.CACHE_DIR / "packed" / catalog._slug(spec) / re.sub(r"[^\w.-]", "_", config) / re.sub(r"[^\w.-]", "_", split) / str(int(index)) | |
| shutil.rmtree(root, ignore_errors=True) | |
| for rel, data in task["files"].items(): | |
| if rel.startswith("solution/"): | |
| continue | |
| dst = (root / rel).resolve() | |
| if not str(dst).startswith(str(root.resolve()) + "/"): | |
| continue # unpack() already drops these; belt and braces | |
| dst.parent.mkdir(parents=True, exist_ok=True) | |
| dst.write_bytes(data) | |
| if rel.endswith(".sh"): | |
| dst.chmod(0o755) | |
| return root | |
| # ── the contract ───────────────────────────────────────────────────────────── | |
| def ref_of(config: str, split: str, i: int) -> str: | |
| """A row's task ref: `<split>/<row>` in the default config, `<config>/<split>/<row>` otherwise.""" | |
| return f"{split}/{i}" if config == "default" else f"{config}/{split}/{i}" | |
| def parse_ref(ref: str) -> tuple[str, str, int]: | |
| parts = [p for p in str(ref).split("/") if p] | |
| if len(parts) < 2 or not parts[-1].isdigit(): | |
| raise LookupError("not a row of this dataset") | |
| return ("/".join(parts[:-2]) or "default"), parts[-2], int(parts[-1]) | |
| def subset_id(config: str, split: str) -> str: | |
| return split if config == "default" else f"{config}/{split}" | |
| def parse_subset(s: str | None) -> tuple[str | None, str | None]: | |
| if not s: | |
| return None, None | |
| return tuple(s.rsplit("/", 1)) if "/" in s else ("default", s) # type: ignore[return-value] | |
| _ICONS = {"task": "message", "trace": "message", "grading": "target", "tests": "flask", "rubric": "list", "environment": "box", | |
| "files": "folder", "metadata": "list", "tools": "wrench", "result": "trophy", "music": "music", "request": "send", | |
| "run": "play", "data": "list", "reward": "target"} | |
| class RowsAdapter(c.Adapter): | |
| """Rows datasets behind the environment contract. One adapter for the family; which row reader (processor) reads a | |
| dataset is picked from its first rows, and names the framework.""" | |
| id = "rows" | |
| name = "Rows" | |
| framework = "Rows" | |
| about = "" | |
| kinds = ("dataset",) | |
| run_note = ("No harness runs this format here yet: its rows hold the task, not the environment to run it in. A row " | |
| "reader can name a runnable form (a Harbor task, packed or converted) to run it.") | |
| def detect(self, kind: str, meta: dict[str, Any]) -> float: | |
| return 0.3 if kind == "dataset" else 0.0 # anything with rows the viewer (or its files) can read | |
| def summary(self, env, subset: str | None = None) -> dict[str, Any]: | |
| config, split = parse_subset(subset) | |
| d = describe(env.id, env.token, config, split, overview=True) | |
| subsets = [{"id": subset_id(x["config"], s), "label": s if x["config"] == "default" else f"{x['config']} · {s}"} | |
| for x in d["configs"] for s in x["splits"]] | |
| roles = {k: v for k, v in d["roles"].items() if v} | |
| return {"state": "ready", "total": d["total"], "inline": False, "search": d["search"], "source": d["source"], | |
| "subsets": subsets, "subset": subset_id(d["config"], d["split"]), | |
| "facets": [c.facet(f["column"], f["column"].replace("_", " "), "level" if re.search(r"difficult|tier|level", f["column"], re.I) else "count", | |
| values=f["values"], sampled=f.get("sampled", False)) for f in d["facets"]], | |
| "how": {"framework": d["processor"]["framework"], "name": d["processor"]["name"], "about": d["processor"]["about"], "roles": roles, | |
| "id": d["processor"]["id"]}, | |
| "overview": d.get("overview") or []} | |
| def tasks(self, env, *, subset=None, q="", filters=None, offset=0, everything=False) -> dict[str, Any]: | |
| config, split = parse_subset(subset) | |
| if config is None: | |
| d = describe(env.id, env.token) | |
| config, split = d["config"], d["split"] | |
| r = rows(env.id, config, split, offset, q, filters or {}, env.token) | |
| cards = [c.card(ref_of(r["config"], r["split"], x["i"]), x["title"], id=x.get("id"), brief=x.get("snippet") or "", chips=x.get("chips") or [], | |
| lead=x.get("lead"), stats=x.get("stats")) | |
| for x in r["rows"]] | |
| return {"cards": cards, "total": r["total"], "offset": r["offset"], "page": r["page"], "note": r.get("note"), "mode": r["mode"], | |
| "subset": subset_id(r["config"], r["split"])} | |
| def task(self, env, ref: str) -> dict[str, Any]: | |
| config, split, i = parse_ref(ref) | |
| v = task(env.id, config, split, i, env.token) | |
| sections = [] | |
| for s in v["sections"]: | |
| k, body = s["kind"], s["body"] | |
| blocks = (list(body or []) if k == "blocks" | |
| else [c.markdown(body if isinstance(body, str) else "")] if k == "markdown" | |
| else [c.messages(body)] if k == "messages" | |
| else [c.code(body.get("text", ""), body.get("path", ""), body.get("path"))] if k == "code" | |
| else [c.files(body.get("tree") or [])] if k == "files" | |
| else [c.value(body)]) | |
| sections.append(c.section(s["id"], s["title"], blocks, icon=_ICONS.get(s["id"], "list"), note=s.get("note") or "")) | |
| total = v.get("total") | |
| sub = subset_id(config, split) | |
| links = [] | |
| twin = v.get("run") | |
| if twin: # the same task in a Harbor dataset (MiMo's conversion): its folder's files are this task's files too | |
| links.append(c.link(f"In Harbor's format: {twin['dataset']}", f"/t/{twin['dataset']}/{twin['path']}", | |
| "the same task, converted; its rollouts are this task's too", rel="same")) | |
| if not any(s["id"] == "files" for s in sections): | |
| try: | |
| from .. import catalog | |
| t = catalog.task(twin["dataset"], twin["path"]) | |
| sections.append(c.section("files", "Files", [c.files(t["tree"], t.get("tree_truncated"), c.hub_folder(twin["dataset"], t["sha"], twin["path"]))], icon="folder", | |
| note=f"its Harbor folder, {twin['path']}")) | |
| except Exception: # noqa: BLE001 - the twin's index isn't here: no files section | |
| pass | |
| return {"ref": ref, "title": v["title"], "id": v.get("id"), "chips": [x for x in v.get("chips") or [] if x], "sections": sections, | |
| "glance": [[a, b] for a, b in v.get("glance") or []], "withheld": v.get("withheld") or [], "links": links, | |
| "framework": v.get("framework") or v["processor"]["framework"], "restricted": v.get("restricted"), | |
| "hub": f"https://huggingface.co/datasets/{env.id}/viewer/{config}/{split}?row={i}", | |
| "nav": {"prev": ref_of(config, split, i - 1) if i > 0 else None, | |
| "next": ref_of(config, split, i + 1) if total is None or i + 1 < total else None, | |
| "index": i, "total": total, "subset": sub}, | |
| "truncated": v.get("truncated") or [], "summary": v.get("summary") or None, | |
| "_packed_run": v.get("packed_run"), "_twin": v.get("run"), "_framework_run": v.get("framework_run")} | |
| def _twin(self, env, ref: str) -> dict[str, Any] | None: | |
| config, split, i = parse_ref(ref) | |
| return task(env.id, config, split, i, env.token).get("run") | |
| def file(self, env, ref: str, path: str) -> dict[str, Any]: | |
| config, split, i = parse_ref(ref) | |
| twin = self._twin(env, ref) | |
| if twin: # its files are its Harbor twin's | |
| from .. import catalog | |
| return catalog.task_file(twin["dataset"], twin["path"], path) | |
| return task_file(env.id, config, split, i, path, env.token) | |
| def folder(self, env, ref: str, path: str) -> dict[str, Any]: | |
| twin = self._twin(env, ref) | |
| if twin: | |
| from .. import catalog | |
| return catalog.task_folder(twin["dataset"], twin["path"], path) | |
| raise LookupError("a row's files are listed up front") | |
| def random(self, env, subset: str | None = None) -> str: | |
| import random as _r | |
| config, split = parse_subset(subset) | |
| d = describe(env.id, env.token, config, split) | |
| return ref_of(d["config"], d["split"], _r.randrange(max(1, d["total"] or 1))) | |
| # running: what the row's reader can name as runnable (a packed Harbor task, a Harbor twin) | |
| def run_options(self, env, ref: str, view: dict[str, Any]) -> list[dict[str, Any]]: | |
| from .. import catalog | |
| from .harbor import harbor_option | |
| if view.get("_packed_run"): | |
| p = view["_packed_run"] | |
| return [harbor_option(p["runnable"], p.get("toml") or "")] | |
| if view.get("_framework_run"): # runs with its own framework (NeMo Gym, Verifiers, verl): the page shows how | |
| return [view["_framework_run"]] | |
| twin = view.get("_twin") | |
| if twin: | |
| try: | |
| t = catalog.task(twin["dataset"], twin["path"]) | |
| except Exception as e: # noqa: BLE001 - its twin isn't readable now | |
| return [c.run_option("harbor", "Harbor, on an HF Sandbox", ok=False, why=f"its Harbor form can't be read ({type(e).__name__})")] | |
| opt = harbor_option(t["runnable"], t.get("toml") or "") | |
| opt["notes"] = [f"Runs `{twin['dataset']}` · `{twin['path']}`: this task in Harbor's format."] + opt["notes"] | |
| return [opt] | |
| return [] | |
| def materialize(self, env, ref: str): | |
| from .. import catalog | |
| config, split, i = parse_ref(ref) | |
| view = self.task(env, ref) | |
| if view.get("_packed_run"): | |
| return materialize(env.id, config, split, i, env.token) | |
| if view.get("_twin"): | |
| return catalog.task_dir(view["_twin"]["dataset"], view["_twin"]["path"]) | |
| raise LookupError("this row has no runnable form") | |
| def run_task(self, env, ref: str) -> dict[str, Any]: | |
| from .. import catalog | |
| config, split, i = parse_ref(ref) | |
| view = self.task(env, ref) | |
| if (view.get("_framework_run") or {}).get("runner") == "nemo-gym": | |
| from .nemogym import agent_of | |
| from .. import nemo_runner | |
| *_, row, truncated = _row(env.id, config, split, i, env.token) | |
| if truncated: | |
| raise ValueError("This task row is truncated; use the complete dataset with NeMo Gym") | |
| agent = agent_of(row, env.id) | |
| nemo_runner.validate(row, agent) | |
| return {"title": view["title"], "restricted": bool(env.meta.get("restricted")), "sha": env.meta.get("sha"), | |
| "native_task": {"row": row, "agent": agent}} | |
| if view.get("_packed_run"): | |
| p = packed(env.id, config, split, i, env.token) | |
| return {"title": p["title"], "restricted": p["restricted"], "sha": p["sha"], "bytes": p["bytes"], "runnable": p["runnable"], | |
| "image": p["image"] or p["runnable"].get("base")} | |
| if view.get("_twin"): | |
| t = catalog.task(view["_twin"]["dataset"], view["_twin"]["path"]) | |
| return {"title": view["title"], "restricted": bool(env.meta.get("restricted")), "sha": env.meta.get("sha"), "bytes": t.get("bytes") or 0, | |
| "runnable": t["runnable"], "image": t["env"]["image"] or t["runnable"].get("base")} | |
| raise LookupError("this row has no runnable form") | |
| def aliases(self, env, ref: str) -> list[tuple[str, str]]: | |
| try: | |
| twin = self.task(env, ref).get("_twin") | |
| except Exception: # noqa: BLE001 | |
| return [] | |
| return [(twin["dataset"], twin["path"])] if twin else [] | |
| def mcp_tools(self, env) -> list[dict[str, Any]]: | |
| return [] | |
| def mcp_call(self, env, name: str, args: dict[str, Any]) -> Any: | |
| raise LookupError(f"no tool named {name!r}") | |