Collapse cohort layer into one SpatialDataCohort; version-pinned freeze/manifest

docs/tutorials/cohorts.md

@@ -57,65 +57,48 @@ Only URLs are replayed, so a store republished at the same URL is not detected.
 Pass `pin_versions=True` to record the catalog version; downloading into a directory pinned to a different
 version then raises `ManifestVersionMismatch` unless `allow_version_change=True`.
 
-## Freeze (network, not executed here)
+## Freeze (offline, not executed here)
 
-`freeze()` pins the selection to its exact members, reads each store's fingerprint, and returns a
-`FrozenCohort` with a deterministic `hash`. `write` saves it as JSON (large member lists go to a parquet
-sidecar); `FrozenCohort.load` restores it offline, without a catalog.
+`freeze()` pins the selection to its exact members (full uids) and returns a `CohortManifest` with a
+deterministic `hash`. `write` saves it as JSON (large member lists go to a parquet sidecar);
+`CohortManifest.load` restores it offline, and `SpatialDataCohort.from_manifest` rebuilds the cohort
+without a catalog query.
 
 ```python
 # not executed
-from sddb import CohortDriftError, FrozenCohort
+from sddb import CohortManifest, SpatialDataCohort
 
-frozen = cohort.freeze()
-frozen.write("cohort.json")
-frozen = FrozenCohort.load("cohort.json")  # offline reload
-frozen.hash, frozen.members
+manifest = cohort.freeze()
+manifest.write("cohort.json")
+manifest = CohortManifest.load("cohort.json")  # offline reload
+cohort = SpatialDataCohort.from_manifest(manifest)
+manifest.hash
 ```
 
-### Verify-on-read
+### Liveness and updates
 
-A `zarr_url` is a "latest" pointer, not a version-pinned one. So reading a frozen cohort's data
-(`frozen.to_anndata()`, `frozen.build(...)`) re-checks each object's fingerprint first and raises
-`CohortDriftError` if a store changed since the freeze. Pass `verify=False` to skip the check.
-A live `SpatialDataCohort.to_anndata()` is unfrozen and performs no such check.
-
-```python
-# not executed
-adata = frozen.to_anndata()  # table=None: default table of each member
-```
+A `zarr_url` is a "latest" pointer, not a version-pinned one. `cohort.verify()` checks that every pinned
+member still resolves, and `cohort.check_updates(catalog)` reports members that a newer catalog version
+supersedes. `download(strict=...)` controls whether a missing pinned store is an error.
 
 ## Split
 
 `split` assigns whole studies (`by="study_id"`) to one fold, so no study leaks across train, val and test.
 It is deterministic for a given `seed`, independent of row order, and refuses to create empty folds on
-tiny cohorts. The `SplitManifest` records the parent cohort hash, and can be written and loaded.
+tiny cohorts. It returns `{fold: SpatialDataCohort}`; each fold carries a `SplitManifest` (pinned to the
+parent cohort hash) as `.split_manifest`, which can be written and loaded.
 
 ```python
 # not executed
 from sddb import SplitManifest
 
-split = frozen.split(by="study_id", train=0.8, val=0.1, test=0.1, seed=42)
-split.write("split.json")
+folds = cohort.split(by="study_id", train=0.8, val=0.1, test=0.1, seed=42)
+folds["train"].split_manifest.write("split.json")
 split = SplitManifest.load("split.json")
 ```
 
-## Check and build with a task adapter
-
-A `TaskAdapter` (protocol in `sddb.adapter`) states what a task needs (`requirements`), validates metadata
-(`validate_meta`) and data (`validate`), and turns a store into a model input (`build`). Concrete adapters
-live in method repos, not in this client; `my_adapter` below stands for one of them.
+## Task adapters
 
-`check` reports compatibility from metadata (add `deep=True` to also open the stores). `build` returns a lazy
-task view; with `materialize=True` each member is built once and provenance is written to `out`. A `split`
-must derive from the same cohort, otherwise `ValueError` is raised.
-
-```python
-# not executed
-report = frozen.check(my_adapter)
-print(report)
-
-view = frozen.build(my_adapter, split=split, materialize=True, out="./task")
-for uid, load in view:
-    sdata = load()
-```
+A `TaskAdapter` (protocol in `sddb.adapter`) states what a task needs and turns a store into a model input.
+Concrete adapters and their execution live in method repos, not in this client. Use `cohort.sdatas` or
+`cohort.load(elements=[...])` to get the data to hand to an adapter.

src/sddb/__init__.py

@@ -7,23 +7,21 @@
 except PackageNotFoundError:  # pragma: no cover
     __version__ = "0.0.0"
 
-from sddb.adapter import PreflightReport, TaskAdapter
+from sddb.adapter import TaskAdapter
 from sddb.catalog import Catalog
 from sddb.cohort import SpatialDataCohort
 from sddb.dataset import Dataset
-from sddb.freeze import CohortDriftError, FrozenCohort
-from sddb.manifest import Manifest, ManifestVersionMismatch
+from sddb.manifest import CohortDriftError, CohortManifest, Manifest, ManifestVersionMismatch
 from sddb.remote import open_sdata
 from sddb.split import SplitManifest
 
 __all__ = [
     "Catalog",
     "CohortDriftError",
+    "CohortManifest",
     "Dataset",
-    "FrozenCohort",
     "Manifest",
     "ManifestVersionMismatch",
-    "PreflightReport",
     "SpatialDataCohort",
     "SplitManifest",
     "TaskAdapter",

src/sddb/_catalog_schema.py

@@ -114,6 +114,8 @@
 COLLECTION_REQUIRED: tuple[str, ...] = ("slug", "name", "members")
 
 FACETS: tuple[str, ...] = (
+    "study_id",
+    "doi",
     "organism",
     "tissue",
     "disease",

src/sddb/adapter.py

@@ -1,21 +1,14 @@
-"""Task-adapter interface: turn a frozen cohort into a task-specific view with provenance."""
+"""Task-adapter interface: the contract a downstream task repo implements.
+
+The client defines only this Protocol. Building/materializing a task view from a cohort lives
+downstream of the cohort, not in the client.
+"""
 
 from __future__ import annotations
 
-import json
-import re
-from collections.abc import Callable, Iterator
-from dataclasses import dataclass
-from datetime import UTC, datetime
-from pathlib import Path
 from typing import Any, Protocol, runtime_checkable
 
-import pandas as pd
-
-import sddb
-from sddb import remote
-from sddb.concat import _table_name
-from sddb.objectmeta import ObjectMeta, fetch_many
+from sddb.objectmeta import ObjectMeta
 
 Issue = str
 
@@ -43,178 +36,3 @@ def validate(self, sdata: Any) -> list[Issue]:
     def build(self, sdata: Any) -> Any:
         """Produce the task-specific item for one object."""
         ...
-
-
-@dataclass
-class PreflightReport:
-    """Counts of compatible members and excluded uids grouped by reason."""
-
-    total: int
-    compatible: int
-    excluded: dict[Issue, list[str]]  # reason -> uids
-    tier: str  # "advisory" | "authoritative"
-
-    def __repr__(self) -> str:
-        lines = [f"{self.total} candidate / {self.compatible} compatible  [{self.tier}]"]
-        lines += [f"  {len(uids)} {reason}: {', '.join(uids)}" for reason, uids in sorted(self.excluded.items())]
-        return "\n".join(lines)
-
-
-_ELEMENT_KINDS = remote._ELEMENT_KINDS
-
-
-def _metas(members: pd.DataFrame, **kw: Any) -> list[ObjectMeta]:
-    return fetch_many([(str(r.uid), str(r.zarr_url)) for r in members.itertuples(index=False)], **kw)
-
-
-def _match(name: str, select: str, how: str) -> bool:
-    """exact/suffix/substring are case-insensitive; regex uses the pattern's own flags."""
-    n, s = name.lower(), select.lower()
-    if how == "exact":
-        return n == s
-    if how == "suffix":  # anchored: `square_016um` must not catch `square_016um_for_vitessce`
-        return n == s or (n.endswith(s) and not n[len(n) - len(s) - 1].isalnum())
-    if how == "substring":
-        return s in n
-    if how == "regex":
-        return re.search(select, name) is not None
-    raise ValueError(f"unknown match mode {how!r}")
-
-
-def _sdata_elements(sdata: Any) -> list[str]:
-    """Sorted ``"kind/name"`` of every element present in an opened SpatialData."""
-    return sorted(f"{kind}/{name}" for kind in _ELEMENT_KINDS for name in getattr(sdata, kind))
-
-
-def _resolve_elements(url: str, roles: list[dict[str, Any]]) -> list[str]:
-    from sddb.remote import elements as _elements
-
-    inv = _elements(url)  # {"kind/name": {...}}
-    resolved: list[str] = []
-    for role in roles:
-        kind = role["role"]
-        kind = kind if kind in _ELEMENT_KINDS else kind + "s"  # image -> images, shape -> shapes, ...
-        select, how = role["select"], role.get("match", "suffix")
-        hits = [p for p in inv if p.split("/", 1)[0] == kind and _match(p.split("/", 1)[1], select, how)]
-        if len(hits) == 1:
-            resolved.append(hits[0])
-        elif not hits:
-            raise ValueError(f"role {role!r}: no element matched; present: {sorted(inv)}")
-        else:
-            raise ValueError(f"role {role!r}: matched {len(hits)} elements {hits}; tighten `select`/`match`")
-    return resolved
-
-
-def _requirements(adapter_obj: Any) -> dict[str, Any]:
-    req = getattr(adapter_obj, "requirements", None)
-    return (req() if callable(req) else None) or {}
-
-
-def _open(url: str, adapter_obj: TaskAdapter) -> Any:
-    from sddb.remote import open_sdata
-
-    roles = _requirements(adapter_obj).get("elements")
-    if not roles:
-        return open_sdata(url, lazy=False)
-    return open_sdata(url, lazy=False, elements=_resolve_elements(url, roles))
-
-
-def check(members: pd.DataFrame, adapter: TaskAdapter, *, deep: bool = False) -> PreflightReport:
-    """Compatibility report. ``deep=False`` uses metadata only (advisory); ``deep=True`` opens each object."""
-    excluded: dict[Issue, list[str]] = {}
-    compatible = 0
-    if deep:
-        pairs = [(str(r.uid), adapter.validate(_open(str(r.zarr_url), adapter))) for r in members.itertuples(index=False)]
-        tier = "authoritative"
-    else:
-        pairs = [(m.uid, adapter.validate_meta(m)) for m in _metas(members)]
-        tier = "advisory"
-    for uid, issues in pairs:
-        if issues:
-            for i in issues:
-                excluded.setdefault(i, []).append(uid)
-        else:
-            compatible += 1
-    return PreflightReport(len(members), compatible, excluded, tier)
-
-
-class TaskView:
-    """Lazy iterable of ``(uid, thunk)``; calling a thunk opens the object and runs ``adapter.build``."""
-
-    def __init__(self, items: list[tuple[str, Callable[[], Any]]]) -> None:
-        self._items = items
-
-    def __iter__(self) -> Iterator[tuple[str, Callable[[], Any]]]:
-        return iter(self._items)
-
-    def __len__(self) -> int:
-        return len(self._items)
-
-    def __getitem__(self, i: int) -> tuple[str, Callable[[], Any]]:
-        return self._items[i]
-
-
-def build(
-    members: pd.DataFrame,
-    adapter: TaskAdapter,
-    *,
-    split: Any = None,
-    materialize: bool = False,
-    out: str | Path | None = None,
-    verify: bool = True,
-    cohort_hash: str = "",
-) -> TaskView:
-    """Task view over ``members``. Lazy by default; ``materialize=True`` builds all and writes provenance.
-
-    With ``materialize=True`` each ``adapter.build(sdata)`` runs once for its side effects and provenance; the
-    returned ``TaskView`` stays lazy (re-invoking a thunk re-opens the store).
-    """
-    if split is not None and split.parent_hash != cohort_hash:
-        raise ValueError(f"split.parent_hash {split.parent_hash!r} does not match cohort hash {cohort_hash!r}")
-    if verify:
-        from sddb.freeze import verify_members
-
-        verify_members(members)
-
-    def thunk(url: str) -> Callable[[], Any]:
-        return lambda: adapter.build(_open(url, adapter))
-
-    view = TaskView([(str(r.uid), thunk(str(r.zarr_url))) for r in members.itertuples(index=False)])
-    if not materialize:
-        return view
-    out_dir = Path(out or ".")
-    out_dir.mkdir(parents=True, exist_ok=True)
-    reqs = _requirements(adapter)
-    roles = reqs.get("elements")
-    produced = []
-    for r in members.itertuples(index=False):
-        sdata = _open(str(r.zarr_url), adapter)
-        out_item = adapter.build(sdata)
-        produced.append(
-            {
-                "uid": str(r.uid),
-                "table": _table_name(r),
-                "output": type(out_item).__name__,
-                "elements": _sdata_elements(sdata) if roles else None,
-            }
-        )
-    split_id = (
-        None
-        if split is None
-        else {"by": split.by, "seed": split.seed, "ratios": split.ratios, "mode": split.mode}
-    )
-    (out_dir / "provenance.json").write_text(
-        json.dumps(
-            {
-                "cohort_hash": cohort_hash,
-                "created_at": datetime.now(UTC).isoformat(),
-                "split": split_id,
-                "adapter": {"name": adapter.name, "version": adapter.version, "config": adapter.config},
-                "sddb_version": sddb.__version__,
-                "requirements": reqs,
-                "members": produced,
-            },
-            indent=2,
-        )
-    )
-    return view

src/sddb/catalog.py

@@ -272,7 +272,7 @@ def search(self, text: str) -> SpatialDataCohort:
         tokens = re.findall(r"\w+", text.lower())
         matched: dict[str, list[str]] = {}
         for col in schema.FACETS:
-            if col not in self._df.columns:
+            if col == "doi" or col not in self._df.columns:  # DOIs tokenize to "10" etc.
                 continue
             values = sorted(str(v) for v in self._df[col].dropna().unique())
             hits = [v for v in values if any(_token_matches(t, v) for t in tokens)]
@@ -289,3 +289,23 @@ def search(self, text: str) -> SpatialDataCohort:
         return SpatialDataCohort(
             self._df[mask.fillna(False).astype(bool)], matched=matched, source=self._source(), filter={"search": text}
         )
+
+    def cohort(self, *, uids: list[str] | None = None, study_ids: list[str] | None = None) -> SpatialDataCohort:
+        """Build a cohort from an explicit set of ``uids`` or ``study_ids`` (exactly one).
+
+        Raises ``KeyError`` naming any id not in the catalog; a valid but empty result warns.
+        """
+        if (uids is None) == (study_ids is None):
+            raise ValueError("pass exactly one of uids= or study_ids=")
+        col, want = ("uid", uids) if uids is not None else ("study_id", study_ids)
+        if col not in self._df.columns:
+            raise ValueError(f"catalog has no {col!r} column")
+        want = [str(x) for x in want]
+        sub = self._df[self._df[col].astype("string").isin(want)]
+        found = set(sub[col].astype("string"))
+        missing = [x for x in want if x not in found]
+        if missing:
+            raise KeyError(f"unknown {col}(s) not in catalog: {missing}")
+        if sub.empty:
+            warnings.warn(f"no catalog rows for {col} in {want}", stacklevel=2)
+        return SpatialDataCohort(sub, source=self._source(), filter={col: sorted(want)})