""" models_client.py — drop-in BESTIARY catalog reader for any PlinyOS app. Copy this single file into any repo that needs the model list. It reads the catalog BESTIARY produces and gives you filtered views, with zero third-party deps. No need to import the full bestiary engine. Resolution order for the catalog: 1. $BESTIARY_CATALOG (explicit path or http(s) URL) 2. a local catalog.json next to this file 3. the canonical monorepo path (…/03-PLINYOS/organs/bestiary/state/catalog.json) Usage ───── from models_client import models, model_ids, newest # frontier / hosted apps (godmode, tempest, libertarium): api_models = models(channel="api") # OpenRouter-reachable ids = model_ids(channel="api", vendor="anthropic") # ["anthropic/claude-opus-4.8", …] # open-weight apps (obliteratus, incanta, pandora): weights = models(open_weight=True) # downloadable, with hf_id coders = models(open_weight=True, capability="tools") # newest drops across everything: for m in newest(days=7): print(m["id"], m["released"]) """ from __future__ import annotations import json import os import urllib.request from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from .service_contracts import ( normalized_text_list, validate_catalog, validate_day_window, ) _HERE = Path(__file__).resolve().parent _CANONICAL = _HERE.parent / "bestiary" / "state" / "catalog.json" _LOCAL = _HERE / "catalog.json" _CHANNEL_ALIASES = {"api": "openrouter", "hosted": "openrouter", "frontier": "openrouter", "hf": "huggingface"} _CATALOG_TIMEOUT_SECONDS = 15 _MAX_CATALOG_BYTES = 8 * 1024 * 1024 def _utc_now() -> datetime: return datetime.now(timezone.utc) def _decode_catalog(payload: bytes, source: str) -> dict[str, Any]: if len(payload) > _MAX_CATALOG_BYTES: raise ValueError(f"BESTIARY catalog exceeds size limit: {source}") try: decoded = payload.decode("utf-8") parsed = json.loads(decoded) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValueError(f"BESTIARY catalog is not valid UTF-8 JSON: {source}") from exc return validate_catalog(parsed) def _read_catalog_path(path: Path) -> dict[str, Any]: try: size = path.stat().st_size except FileNotFoundError: raise FileNotFoundError(f"BESTIARY catalog not found: {path}") from None if size > _MAX_CATALOG_BYTES: raise ValueError(f"BESTIARY catalog exceeds size limit: {path}") return _decode_catalog(path.read_bytes(), str(path)) def _load() -> dict[str, Any]: src = os.environ.get("BESTIARY_CATALOG", "").strip() if src.startswith("http://") or src.startswith("https://"): with urllib.request.urlopen(src, timeout=_CATALOG_TIMEOUT_SECONDS) as response: return _decode_catalog(response.read(_MAX_CATALOG_BYTES + 1), src) if src: return _read_catalog_path(Path(src)) for path in (_LOCAL, _CANONICAL, _HERE / "state" / "catalog.json"): if path.exists(): return _read_catalog_path(path) raise FileNotFoundError( "BESTIARY catalog not found. Run `python3 bestiary.py update`, or set " "$BESTIARY_CATALOG to a catalog.json path or URL." ) def _normalize_filter(value: str | None, name: str) -> str | None: if value is None: return None if not isinstance(value, str): raise TypeError(f"{name} must be a string") normalized = value.strip().casefold() return normalized or None def models(channel: str | None = None, vendor: str | None = None, open_weight: bool | None = None, capability: str | None = None) -> list[dict]: """Return catalog records filtered by channel / vendor / open_weight / capability.""" channel = _normalize_filter(channel, "channel") vendor = _normalize_filter(vendor, "vendor") capability = _normalize_filter(capability, "capability") if open_weight is not None and not isinstance(open_weight, bool): raise TypeError("open_weight must be a boolean") if channel in ("open-weight", "openweight", "weights"): open_weight, channel = True, None if channel: channel = _CHANNEL_ALIASES.get(channel, channel) out = [] for m in _load().get("models", []): channels = normalized_text_list(m, "channels") capabilities = normalized_text_list(m, "capabilities") record_vendor = m.get("vendor") if record_vendor is not None and not isinstance(record_vendor, str): raise ValueError("BESTIARY catalog vendor must be a string") record_open_weight = m.get("open_weight", False) if not isinstance(record_open_weight, bool): raise ValueError("BESTIARY catalog open_weight must be a boolean") if channel and channel not in channels: continue if vendor and (record_vendor or "").strip().casefold() != vendor: continue if open_weight is not None and record_open_weight != open_weight: continue if capability and capability not in capabilities: continue out.append(m) return out def model_ids(**kw) -> list[str]: """Just the ids — the common case for populating a dropdown or a config.""" ids = [] for record in models(**kw): model_id = record.get("id") if not isinstance(model_id, str) or not model_id.strip(): raise ValueError("BESTIARY catalog model requires a non-empty string id") ids.append(model_id) return ids def newest(days: int = 7, **kw) -> list[dict]: """Models first-seen (≈ released) within the last `days`, newest first.""" cutoff = _utc_now() - timedelta(days=validate_day_window(days)) out: list[tuple[datetime, dict]] = [] for m in models(**kw): fs = m.get("first_seen") or m.get("released") if not fs: continue if not isinstance(fs, str): raise ValueError("BESTIARY catalog dates must be strings") try: t = datetime.strptime(fs, "%Y-%m-%d").replace(tzinfo=timezone.utc) except ValueError: continue if t >= cutoff: out.append((t, m)) out.sort(key=lambda item: item[0], reverse=True) return [record for _, record in out] if __name__ == "__main__": cat = _load() print(f"catalog generated_at={cat.get('generated_at')} count={cat.get('count')}") print(f" api models : {len(models(channel='api'))}") print(f" open-weight : {len(models(open_weight=True))}") print(f" new (7d) : {len(newest(7))}")