| """The Hugging Face extension — the Hub as mindXtrain needs it. |
| |
| Design rules, learned the hard way on a live node (2026-09): |
| |
| - **Membership is not write scope.** `whoami` lists orgs you belong to; only the token's own |
| permissions say where it can write. `account()` reports both, separately. |
| - **`list_repo_tree` yields RepoFile AND RepoFolder**, both carrying `.path`; only |
| `repo_info().siblings` carry `.rfilename`. Reading the wrong one makes a scan report "nothing |
| on the Hub" while the repo is full. |
| - **The Hub checks a Space's ZeroGPU quota before it checks existence**: `create_repo(exist_ok=True, |
| space_hardware=…)` on an EXISTING Space answers 402 once the account's slots are used. Check |
| existence first. |
| - **A Space README's `short_description` must be ≤ 60 characters** or the upload is refused. |
| - **Never park a write-scoped token on a public Space.** Public Spaces read public repos. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
| import re |
| import time |
| from pathlib import Path |
| from typing import Any |
|
|
| _HF_ENV = ("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACEHUB_API_TOKEN") |
|
|
|
|
| def _token(explicit: str | None = None) -> str | None: |
| if explicit: |
| return explicit |
| for k in _HF_ENV: |
| v = os.environ.get(k) |
| if v: |
| return v |
| return None |
|
|
|
|
| def _api(token: str | None = None): |
| """(HfApi, None) or (None, {"ok": False, …}) — never raises.""" |
| try: |
| from huggingface_hub import HfApi |
| except ImportError: |
| return None, {"ok": False, "reason": "huggingface_hub not installed — `uv sync --extra hf`"} |
| tok = _token(token) |
| if not tok: |
| return None, {"ok": False, "reason": f"no token — set one of {', '.join(_HF_ENV)}"} |
| return HfApi(token=tok), None |
|
|
|
|
| def tree_paths(api, repo_id: str, **kw) -> list[str]: |
| """File paths of a repo tree (folders skipped). See the module docstring: the tree's entries |
| carry `.path`, not `.rfilename`.""" |
| return [f.path for f in api.list_repo_tree(repo_id, **kw) if type(f).__name__ != "RepoFolder"] |
|
|
|
|
| |
| def account(token: str | None = None) -> dict[str, Any]: |
| """The identity behind the token, its role, the orgs it belongs to, and — separately — the |
| namespaces it can actually write.""" |
| api, err = _api(token) |
| if err: |
| return err |
| try: |
| me = api.whoami() |
| except Exception as e: |
| return {"ok": False, "reason": f"whoami failed: {type(e).__name__}: {str(e)[:200]}"} |
| auth = (me.get("auth") or {}).get("accessToken") or {} |
| perms = auth.get("fineGrained") or {} |
| writable, creatable = [], [] |
| for scope in (perms.get("scoped") or []): |
| ent = (scope.get("entity") or {}) |
| name = ent.get("name") or ent.get("type") |
| perms_list = scope.get("permissions") or [] |
| if any(p.startswith("repo.write") or p == "repo.content.write" for p in perms_list): |
| writable.append(name) |
| if "repo.write" in perms_list: |
| creatable.append(name) |
| return {"ok": True, "user": me.get("name"), "type": me.get("type"), "is_pro": bool(me.get("isPro")), |
| "role": auth.get("role"), "orgs": [o.get("name") for o in (me.get("orgs") or [])], |
| "writable_namespaces": writable or None, "creatable_namespaces": creatable or None, |
| "note": "membership is not write scope — a token can belong to an org and still be unable to write it"} |
|
|
|
|
| |
| def pull_base(model_id: str, *, token: str | None = None, allow_patterns: list[str] | None = None) -> dict[str, Any]: |
| """Download a base model to the local cache. Do this BEFORE training: a run that discovers a |
| missing base three hours in has wasted three hours.""" |
| try: |
| from huggingface_hub import snapshot_download |
| except ImportError: |
| return {"ok": False, "reason": "huggingface_hub not installed — `uv sync --extra hf`"} |
| t0 = time.time() |
| try: |
| path = snapshot_download(model_id, token=_token(token), allow_patterns=allow_patterns) |
| except Exception as e: |
| return {"ok": False, "model": model_id, "reason": f"{type(e).__name__}: {str(e)[:200]}"} |
| p = Path(path) |
| return {"ok": True, "model": model_id, "path": str(p), "seconds": round(time.time() - t0, 1), |
| "bytes": sum(f.stat().st_size for f in p.rglob("*") if f.is_file())} |
|
|
|
|
| def warm(config: Path | str, *, token: str | None = None) -> dict[str, Any]: |
| """Pull whatever a run.yaml says it needs (the base model) so `train` starts cold-free.""" |
| try: |
| import yaml |
| cfg = yaml.safe_load(Path(config).read_text()) |
| except Exception as e: |
| return {"ok": False, "reason": f"unreadable config: {type(e).__name__}: {str(e)[:160]}"} |
| base = ((cfg or {}).get("model") or {}).get("name") |
| if not base: |
| return {"ok": False, "reason": "config has no model.name"} |
| return dict(pull_base(base, token=token), config=str(config)) |
|
|
|
|
| |
| def _card(repo_id: str, meta: dict[str, Any]) -> str: |
| """A model card written from the run's own numbers. No claim that is not in `meta`.""" |
| m = meta |
| fm = {"license": m.get("license", "apache-2.0"), "library_name": "transformers", |
| "pipeline_tag": "text-generation", "tags": ["mindxtrain", "lora", *list(m.get("tags") or [])]} |
| if m.get("base"): |
| fm["base_model"] = m["base"] |
| if m.get("dataset"): |
| fm["datasets"] = [m["dataset"]] |
| head = "---\n" + "\n".join( |
| f"{k}: {json.dumps(v) if isinstance(v, (list, dict)) else v}" for k, v in fm.items()) + "\n---\n\n" |
| rows = [("base", m.get("base")), ("method", m.get("method")), ("corpus", m.get("dataset")), |
| ("hardware", m.get("hardware")), ("wall", m.get("wall")), ("train loss", m.get("train_loss")), |
| ("eval loss", m.get("eval_loss")), ("gate", m.get("gate")), ("framework", "mindXtrain " + str(m.get("framework_version") or ""))] |
| table = "\n".join(f"| {k} | {v} |" for k, v in rows if v not in (None, "")) |
| return (head + f"# {repo_id.split('/')[-1]}\n\n{m.get('summary', 'A mindXtrain run, published with its evidence.')}\n\n" |
| f"| | |\n|---|---|\n{table}\n\n" |
| "## Use\n\n```python\nfrom transformers import AutoModelForCausalLM, AutoTokenizer\n" |
| f'tok = AutoTokenizer.from_pretrained("{repo_id}")\nm = AutoModelForCausalLM.from_pretrained("{repo_id}")\n```\n\n' |
| + (f"## The gate\n\n{m['gate_note']}\n\n" if m.get("gate_note") else "") |
| + "## Provenance\n\nTrained by [mindXtrain](https://github.com/professor-codephreak/mindXtrain); " |
| "the training log ships beside the weights.\n") |
|
|
|
|
| def published_modelfile(persona_system: str | None = None, *, |
| repeat_penalty: float | None = 1.3) -> str: |
| """The Modelfile shipped beside a published run. `repeat_penalty=None` leaves the penalty out, |
| so engines that refuse penalties (bankml) load the same file.""" |
| return ("# ollama create <name> -f Modelfile (from this repo's directory)\nFROM .\n" |
| + (f'SYSTEM """{persona_system}"""\n' if persona_system else "") |
| + "PARAMETER temperature 0.7\n" |
| + (f"PARAMETER repeat_penalty {repeat_penalty}\n" if repeat_penalty is not None else "") |
| + 'PARAMETER stop "<|im_end|>"\n') |
|
|
|
|
| def publish_generation(run_dir: Path | str, repo_id: str, *, token: str | None = None, private: bool = False, |
| meta: dict[str, Any] | None = None, persona_system: str | None = None, |
| include_merged: bool = True, dry_run: bool = False, |
| repeat_penalty: float | None = 1.3) -> dict[str, Any]: |
| """A finished run as a model repo: merged weights at the root (if present), the LoRA delta under |
| `adapter/`, `train.log`, a `Modelfile` for Ollama, and a card built from `meta`. |
| |
| `dry_run=True` reports exactly what would be uploaded and touches nothing. |
| |
| `repeat_penalty` is written into the Modelfile (default 1.3, the imprint gate's decoding); |
| `None` leaves it out, so the same Modelfile also loads in engines that refuse penalties |
| (bankml reproduces temperature/top-k/top-p/min-p, not penalties).""" |
| run = Path(run_dir) |
| if not run.is_dir(): |
| return {"ok": False, "reason": f"no run dir at {run}"} |
| ck = run / "checkpoint" |
| merged = run / "ollama_push" / "merged" |
| staged: dict[str, Path] = {} |
| if include_merged and merged.is_dir(): |
| for f in merged.iterdir(): |
| if f.is_file(): |
| staged[f.name] = f |
| if ck.is_dir(): |
| for f in ck.iterdir(): |
| if (f.is_file() and f.name != "training_args.bin") or f.name == "training_args.bin": |
| staged[f"adapter/{f.name}"] = f |
| log = run / "train.log" |
| if log.is_file(): |
| staged["train.log"] = log |
| if not staged: |
| return {"ok": False, "reason": f"{run} holds neither a checkpoint nor merged weights"} |
| meta = dict(meta or {}) |
| if not meta.get("base"): |
| try: |
| meta["base"] = json.loads((ck / "adapter_config.json").read_text()).get("base_model_name_or_path") |
| except Exception: |
| pass |
| card = _card(repo_id, meta) |
| modelfile = published_modelfile(persona_system, repeat_penalty=repeat_penalty) |
| plan = {"repo": repo_id, "private": private, "files": [*sorted(staged), "README.md", "Modelfile"], |
| "bytes": sum(f.stat().st_size for f in staged.values())} |
| if dry_run: |
| return {"ok": True, "dry_run": True, "would_upload": plan, "card_preview": card[:400]} |
| api, err = _api(token) |
| if err: |
| return err |
| import shutil |
| import tempfile |
| with tempfile.TemporaryDirectory() as tmp: |
| stage = Path(tmp) |
| for rel, src in staged.items(): |
| dst = stage / rel |
| dst.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(src, dst) |
| (stage / "README.md").write_text(card) |
| (stage / "Modelfile").write_text(modelfile) |
| try: |
| api.create_repo(repo_id, repo_type="model", private=private, exist_ok=True) |
| ci = api.upload_folder(folder_path=str(stage), repo_id=repo_id, repo_type="model", |
| commit_message=meta.get("commit_message") or "mindXtrain: a run, published with its evidence") |
| except Exception as e: |
| return {"ok": False, "repo": repo_id, "reason": f"{type(e).__name__}: {str(e)[:300]}"} |
| return {"ok": True, "repo": repo_id, "url": f"https://huggingface.co/{repo_id}", |
| "commit": str(getattr(ci, "oid", ci))[:12], "uploaded": plan} |
|
|
|
|
| |
| def push_dataset(path: Path | str, repo_id: str, *, token: str | None = None, private: bool = False, |
| path_in_repo: str = "", manifest: dict[str, Any] | None = None) -> dict[str, Any]: |
| """Push a training corpus (a folder or one JSONL) with an optional manifest beside it.""" |
| api, err = _api(token) |
| if err: |
| return err |
| src = Path(path) |
| if not src.exists(): |
| return {"ok": False, "reason": f"no corpus at {src}"} |
| try: |
| api.create_repo(repo_id, repo_type="dataset", private=private, exist_ok=True) |
| if src.is_dir(): |
| ci = api.upload_folder(folder_path=str(src), repo_id=repo_id, repo_type="dataset", |
| path_in_repo=path_in_repo or None, commit_message="mindXtrain: corpus") |
| else: |
| ci = api.upload_file(path_or_fileobj=str(src), path_in_repo=f"{path_in_repo}/{src.name}".lstrip("/"), |
| repo_id=repo_id, repo_type="dataset", commit_message="mindXtrain: corpus") |
| if manifest: |
| api.upload_file(path_or_fileobj=json.dumps(manifest, indent=1).encode(), |
| path_in_repo=f"{path_in_repo}/MANIFEST.json".lstrip("/"), |
| repo_id=repo_id, repo_type="dataset", commit_message="mindXtrain: corpus manifest") |
| except Exception as e: |
| return {"ok": False, "repo": repo_id, "reason": f"{type(e).__name__}: {str(e)[:300]}"} |
| return {"ok": True, "repo": repo_id, "url": f"https://huggingface.co/datasets/{repo_id}", |
| "commit": str(getattr(ci, "oid", ci))[:12]} |
|
|
|
|
| |
| _GEN = re.compile(r"(?:^|/)gen(\d+)(?:/|$)") |
|
|
|
|
| def lineage(repo_id: str, *, repo_type: str = "model", token: str | None = None, |
| local_runs: Path | str | None = None) -> dict[str, Any]: |
| """Generations present in a repo, and — when `local_runs` is given — which local runs are not |
| published yet. Uses `tree_paths`, so folders never break the scan.""" |
| api, err = _api(token) |
| if err: |
| return err |
| try: |
| paths = tree_paths(api, repo_id, repo_type=repo_type, recursive=True) |
| except Exception as e: |
| return {"ok": False, "repo": repo_id, "reason": f"{type(e).__name__}: {str(e)[:200]}"} |
| gens: dict[int, list[str]] = {} |
| for p in paths: |
| m = _GEN.search(p) |
| if m: |
| gens.setdefault(int(m.group(1)), []).append(p) |
| out = {"ok": True, "repo": repo_id, "repo_type": repo_type, "files": len(paths), |
| "generations": {str(g): {"files": len(v), |
| "adapter": any(x.endswith("adapter_model.safetensors") for x in v), |
| "merged": any(x.endswith("model.safetensors") and "adapter" not in x for x in v)} |
| for g, v in sorted(gens.items())}} |
| if local_runs: |
| root = Path(local_runs) |
| local = sorted(int(m.group(1)) for d in root.glob("gen*") if d.is_dir() for m in [_GEN.search(d.name + "/")] if m) |
| out["local"] = local |
| out["unpublished"] = [g for g in local if g not in gens] |
| return out |
|
|
|
|
| |
| def push_space(folder: Path | str, space_id: str, *, token: str | None = None, private: bool = True, |
| hardware: str | None = "zero-a10g", variables: dict[str, str] | None = None, |
| secrets: dict[str, str] | None = None) -> dict[str, Any]: |
| """Push a Gradio folder as a Space. Existence is checked BEFORE creation (the Hub tests the |
| ZeroGPU quota first and answers 402 on a repo that already exists), and a README |
| `short_description` longer than 60 characters is refused by the Hub, so it is checked here.""" |
| api, err = _api(token) |
| if err: |
| return err |
| src = Path(folder) |
| if not (src / "app.py").is_file(): |
| return {"ok": False, "reason": f"{src} has no app.py"} |
| readme = src / "README.md" |
| if readme.is_file(): |
| m = re.search(r"^short_description:\s*(.+)$", readme.read_text(), re.M) |
| if m and len(m.group(1).strip()) > 60: |
| return {"ok": False, "reason": f"short_description is {len(m.group(1).strip())} chars; the Hub allows 60"} |
| try: |
| exists = api.repo_exists(space_id, repo_type="space") |
| if not exists: |
| api.create_repo(space_id, repo_type="space", space_sdk="gradio", private=private, |
| **({"space_hardware": hardware} if hardware else {})) |
| ci = api.upload_folder(folder_path=str(src), repo_id=space_id, repo_type="space", |
| ignore_patterns=["__pycache__/*", "**/__pycache__/*", "*.pyc"], |
| commit_message="mindXtrain UI") |
| for k, v in (variables or {}).items(): |
| api.add_space_variable(space_id, k, v) |
| for k, v in (secrets or {}).items(): |
| api.add_space_secret(space_id, k, v) |
| rt = api.get_space_runtime(space_id) |
| except Exception as e: |
| return {"ok": False, "space": space_id, "reason": f"{type(e).__name__}: {str(e)[:300]}"} |
| return {"ok": True, "space": space_id, "existed": exists, "stage": str(rt.stage), |
| "commit": str(getattr(ci, "oid", ci))[:12], |
| "url": f"https://huggingface.co/spaces/{space_id}", |
| "host": "https://" + space_id.replace("/", "-").replace("_", "-").lower() + ".hf.space", |
| "note": "free personal accounts host 2 ZeroGPU Spaces; a free org hosts none (402)"} |
|
|
