> ## Documentation Index
> Fetch the complete documentation index at: https://docs.bioptimus.com/llms.txt
> Use this file to discover all available pages before exploring further.

# cohort

Cohort manifest for pairing WSIs with bulk RNA and clinical metadata.

Provides `Cohort`, a typed registry that tracks the mapping between patients, wsis, bulk RNA samples, timepoints, labels, and arbitrary clinical metadata.  `WSIRecord` is the single source of truth for each WSI — including per-model outputs and processing status.

Construction paths:

1. **From a user-provided CSV** via `from_csv`.
2. **Auto-matching from directories** via `from_directories`.
3. **From a YAML manifest** via `load` (for resume).

The cohort can be serialised to YAML via `save` and loaded back for full reproducibility and resume support.

**Example:**

```python theme={null}
from bioptimus.data.cohort import Cohort

# Auto-match from directories:
cohort = Cohort.from_directories(
    wsi_dir="/data/wsis",
    bulk_rna_dir="/data/rna",
)

# Or load a saved manifest (resume):
cohort = Cohort.load(".output/brca/run_1/manifest.yaml")

# Add labels post-hoc:
cohort.add_labels({"patient_001": {"mutation": "BRAF"}})

# Iterate:
for record in cohort:
    print(record.patient_id, record.wsi_path)
```

## ModelOutput

```python theme={null}
@dataclass
class ModelOutput()
```

Per-model output record for a single WSI.

<ParamField body="model_name">
  Name of the model that produced this output.
</ParamField>

<ParamField body="embedding_path">
  Path to the embedding output file.
</ParamField>

<ParamField body="prediction_path">
  Path to the prediction output file.
</ParamField>

<ParamField body="tiles_csv_path">
  Path to the tile coordinates CSV.
</ParamField>

<ParamField body="tiles_dir">
  Directory containing exported tile images.
</ParamField>

<ParamField body="modalities">
  Input modalities used for this output (e.g. `["image"]` or `["image", "bulk_rna"]`).
</ParamField>

<ParamField body="timestamp">
  ISO 8601 string of when the output was produced.
</ParamField>

<ParamField body="status">
  Processing status (`"pending"`, `"done"`, `"error"`).
</ParamField>

<ParamField body="error">
  Error message if status is `"error"`.
</ParamField>

#### is\_done

```python theme={null}
@property
def is_done() -> bool
```

Returns `True` if status is done.

#### has\_embedding

```python theme={null}
@property
def has_embedding() -> bool
```

Returns `True` if an embedding path is set and exists.

#### has\_prediction

```python theme={null}
@property
def has_prediction() -> bool
```

Returns `True` if a prediction path is set and exists.

#### set\_stage\_output

```python theme={null}
def set_stage_output(stage: str,
                     modalities: list[str],
                     path: Path,
                     timestamp: str | None = None) -> None
```

Records an output for a specific stage and modality combo.

Updates both the per-modality tracking dict and the top-level convenience fields (`embedding_path` / `prediction_path`).

<ParamField body="stage" type="str" required>
  `"embed"` or `"predict"`.
</ParamField>

<ParamField body="modalities" type="list[str]" required>
  Modalities used for this output.
</ParamField>

<ParamField body="path" type="Path" required>
  Output file path.
</ParamField>

<ParamField body="timestamp" type="str | None">
  ISO 8601 timestamp.
</ParamField>

#### get\_stage\_path

```python theme={null}
def get_stage_path(stage: str,
                   modalities: list[str] | None = None) -> Path | None
```

Returns the output path for a stage and modality combo.

<ParamField body="stage" type="str" required>
  `"embed"` or `"predict"`.
</ParamField>

<ParamField body="modalities" type="list[str] | None">
  Modality combination to look up. Falls back to the top-level path when `None`.
</ParamField>

<ResponseField name="returns" type="Path | None">
  Resolved path, or `None` if not recorded.
</ResponseField>

#### to\_dict

```python theme={null}
def to_dict() -> dict[str, Any]
```

Serialises to a plain dict for YAML output.

<ResponseField name="returns" type="dict[str, Any]">
  A dictionary representation of the model output.
</ResponseField>

#### from\_dict

```python theme={null}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ModelOutput
```

Constructs from a plain dict (YAML round-trip).

<ParamField body="data" type="dict[str, Any]" required>
  Dictionary with serialised model output fields.
</ParamField>

<ResponseField name="returns" type="ModelOutput">
  A new `ModelOutput` instance.
</ResponseField>

## WSIRecord

```python theme={null}
@dataclass
class WSIRecord()
```

Single WSI entry — the single source of truth.

Tracks inputs (WSI, RNA, mask), per-model outputs, and arbitrary annotations for one WSI within a cohort.

<ParamField body="patient_id">
  Unique patient identifier.
</ParamField>

<ParamField body="wsi_id">
  Unique WSI identifier (typically the filename stem).
</ParamField>

<ParamField body="wsi_path">
  Path to the WSI file.
</ParamField>

<ParamField body="bulk_rna_id">
  Identifier for the paired bulk RNA sample.
</ParamField>

<ParamField body="bulk_rna_path">
  Path to the bulk RNA CSV/TSV file.
</ParamField>

<ParamField body="mask_path">
  Path to a tissue mask (pre-computed or cached).
</ParamField>

<ParamField body="mask_timestamp">
  When the mask was computed/discovered.
</ParamField>

<ParamField body="timepoint">
  Temporal ordering label (e.g. `"t0"`, `"t1"`).
</ParamField>

<ParamField body="outputs">
  Per-model output records keyed by model name.
</ParamField>

<ParamField body="labels">
  Arbitrary categorical annotations.
</ParamField>

<ParamField body="metadata">
  Arbitrary clinical/treatment metadata.
</ParamField>

#### get\_output

```python theme={null}
def get_output(model_name: str) -> ModelOutput
```

Returns the output for *model\_name*, creating if absent.

<ParamField body="model_name" type="str" required>
  Model identifier string.
</ParamField>

<ResponseField name="returns" type="ModelOutput">
  The `ModelOutput` for the given model.
</ResponseField>

#### is\_stage\_done

```python theme={null}
def is_stage_done(model_name: str, stage: str) -> bool
```

Checks whether a stage output exists on disk.

When `modality_outputs` are tracked, the check is scoped to the record's current `available_modalities` so that linking new data (e.g. bulk RNA) automatically surfaces pending work without requiring `force=True`.

<ParamField body="model_name" type="str" required>
  Model identifier.
</ParamField>

<ParamField body="stage" type="str" required>
  `"embed"` or `"predict"`.
</ParamField>

<ResponseField name="returns" type="bool">
  `True` if the output path exists on disk.
</ResponseField>

#### has\_mask

```python theme={null}
@property
def has_mask() -> bool
```

Returns `True` if a mask path is set and exists on disk.

#### available\_modalities

```python theme={null}
@property
def available_modalities() -> list[str]
```

Returns the input modalities available for this record.

Always includes `"image"`.  Includes `"bulk_rna"` when `bulk_rna_path` is set.

#### to\_dict

```python theme={null}
def to_dict() -> dict[str, Any]
```

Serialises to a plain dict for YAML output.

<ResponseField name="returns" type="dict[str, Any]">
  A dictionary representation of the WSI record.
</ResponseField>

#### from\_dict

```python theme={null}
@classmethod
def from_dict(cls, data: dict[str, Any]) -> WSIRecord
```

Constructs from a plain dict (YAML round-trip).

<ParamField body="data" type="dict[str, Any]" required>
  Dictionary with serialised WSI record fields.
</ParamField>

<ResponseField name="returns" type="WSIRecord">
  A new `WSIRecord` instance.
</ResponseField>

## PatientRecord

```python theme={null}
@dataclass
class PatientRecord()
```

Groups all WSIs belonging to a single patient.

<ParamField body="patient_id">
  Unique patient identifier.
</ParamField>

<ParamField body="wsis">
  Ordered list of WSI records (by timepoint).
</ParamField>

<ParamField body="labels">
  Patient-level categorical annotations.
</ParamField>

<ParamField body="metadata">
  Patient-level clinical metadata.
</ParamField>

## Cohort

```python theme={null}
class Cohort()
```

Typed cohort manifest — single source of truth for an experiment.

A cohort is an ordered collection of `WSIRecord` entries grouped by patient.  It stores all information needed to reproduce and resume an inference run: file locations, pairing logic, timepoints, per-model outputs, labels, and clinical metadata.

Construction: Use `from_csv`, `from_directories`, or `load` rather than calling the constructor directly.

<ParamField body="records">
  Initial list of `WSIRecord` entries.
</ParamField>

#### from\_csv

```python theme={null}
@classmethod
def from_csv(cls,
             path: str | Path,
             *,
             wsi_dir: str | Path | None = None,
             bulk_rna_dir: str | Path | None = None,
             columns: dict[str, str] | None = None) -> Cohort
```

Creates a cohort from a user-provided CSV manifest.

Required columns:

* `wsi_id`: WSI identifier (filename stem or full name).

Optional columns:

* `patient_id`: if absent, derived from `wsi_id`.
* `bulk_rna_id`: paired RNA sample identifier.
* `timepoint`: temporal label.
* `wsi_path`: explicit path override.
* `bulk_rna_path`: explicit path override.
* Any other columns are treated as labels if prefixed with `label_` or as metadata otherwise.

When *wsi\_dir* or *bulk\_rna\_dir* are provided, paths are resolved by joining the directory with the corresponding ID (plus matching extension found on disk).

The *columns* dict maps canonical field names to the actual CSV column names.  Only the keys that differ from the defaults need to be specified. For example:

```python theme={null}
Cohort.from_csv(
    "cohort.csv",
    columns={
        "wsi_id": "imageid",
        "patient_id": "subject_id",
    },
)
```

Recognised canonical keys: `wsi_id`, `patient_id`, `bulk_rna_id`, `timepoint`, `wsi_path`, `bulk_rna_path`.

<ParamField body="path">
  Path to the CSV manifest.
</ParamField>

<ParamField body="wsi_dir">
  Directory to resolve WSI paths from.
</ParamField>

<ParamField body="bulk_rna_dir">
  Directory to resolve bulk RNA paths from.
</ParamField>

<ParamField body="columns">
  Mapping of canonical field names to actual CSV column names. Unmapped fields fall back to their canonical name.
</ParamField>

**Returns**:

A populated `Cohort`.

#### from\_directories

```python theme={null}
@classmethod
def from_directories(
        cls,
        wsi_dir: str | Path,
        bulk_rna_dir: str | Path | None = None,
        *,
        mask_dir: str | Path | None = None,
        patient_id_fn: Callable[[str], str] | None = None) -> Cohort
```

Creates a cohort by auto-matching files from directories.

WSI and bulk RNA files are paired by **exact filename stem** equality (case-sensitive).  Patient IDs default to the shared stem.  When multiple wsis share the same patient ID, they are assigned sequential timepoints (`t0`, `t1`, ...) in alphabetical order.

<ParamField body="wsi_dir" type="str | Path" required>
  Directory containing WSI files.
</ParamField>

<ParamField body="bulk_rna_dir" type="str | Path | None">
  Directory containing bulk RNA files. When `None`, wsis are registered without RNA.
</ParamField>

<ParamField body="mask_dir" type="str | Path | None">
  Directory containing pre-computed masks (PNG files whose stem matches the WSI stem).
</ParamField>

<ParamField body="patient_id_fn" type="Callable[[str], str] | None">
  Optional callable that maps a filename stem to a patient ID.
</ParamField>

<ResponseField name="returns" type="Cohort">
  A populated `Cohort`.
</ResponseField>

#### save

```python theme={null}
def save(path: str | Path) -> Path
```

Persists the cohort manifest to a YAML file.

The YAML includes all inputs, per-model outputs, status, labels, and metadata — everything needed to resume.

<ParamField body="path" type="str | Path" required>
  Destination file path.
</ParamField>

<ResponseField name="returns" type="Path">
  The resolved output path.
</ResponseField>

#### load

```python theme={null}
@classmethod
def load(cls, path: str | Path) -> Cohort
```

Loads a cohort from a previously saved YAML manifest.

<ParamField body="path" type="str | Path" required>
  Path to the YAML manifest file.
</ParamField>

<ResponseField name="returns" type="Cohort">
  A fully-populated `Cohort`.
</ResponseField>

#### upsert

```python theme={null}
def upsert(wsi_id: str,
           *,
           patient_id: str | None = None,
           wsi_path: Path | None = None,
           bulk_rna_path: Path | None = None,
           mask_path: Path | None = None,
           timepoint: str | None = None) -> WSIRecord
```

Inserts or updates a WSI record.

If a record with *wsi\_id* exists, updates the non-None fields.  Otherwise creates a new record.

<ParamField body="wsi_id" type="str" required>
  WSI identifier.
</ParamField>

<ParamField body="patient_id" type="str | None">
  Patient identifier (defaults to wsi\_id).
</ParamField>

<ParamField body="wsi_path" type="Path | None">
  Path to WSI file.
</ParamField>

<ParamField body="bulk_rna_path" type="Path | None">
  Path to bulk RNA file.
</ParamField>

<ParamField body="mask_path" type="Path | None">
  Path to tissue mask.
</ParamField>

<ParamField body="timepoint" type="str | None">
  Timepoint label.
</ParamField>

<ResponseField name="returns" type="WSIRecord">
  The inserted or updated `WSIRecord`.
</ResponseField>

#### get\_wsi

```python theme={null}
def get_wsi(wsi_id: str) -> WSIRecord | None
```

Returns the record for *wsi\_id*, or `None`.

<ParamField body="wsi_id" type="str" required>
  Unique WSI identifier (typically the file stem).
</ParamField>

<ResponseField name="returns" type="WSIRecord | None">
  The matching `WSIRecord`, or `None` if not found.
</ResponseField>

#### link\_bulk\_rna

```python theme={null}
def link_bulk_rna(bulk_rna_dir: str | Path,
                  *,
                  extensions: set[str] | None = None,
                  separator: str | None = None,
                  gene_column: str | None = None,
                  value_column: str | None = None,
                  strip_version: bool = False) -> int
```

Links bulk RNA files to existing WSI records by stem.

Scans *bulk\_rna\_dir* for recognised RNA files and pairs them with existing records by exact filename-stem equality. Records that already have a `bulk_rna_path` are skipped.

This enables late-binding of bulk RNA data: build a cohort from WSIs first, then call this method to attach RNA when it becomes available.

<ParamField body="bulk_rna_dir" type="str | Path" required>
  Directory containing bulk RNA files.
</ParamField>

<ParamField body="extensions" type="set[str] | None">
  File extensions to match. Defaults to `{".csv", ".tsv"}`.
</ParamField>

<ParamField body="separator" type="str | None">
  Column delimiter for parsing. When `None` the separator is inferred from the file extension.
</ParamField>

<ParamField body="gene_column" type="str | None">
  Column name containing gene identifiers. Required (with `value_column`) for long-format files (e.g. GDC/TCGA gene quantification TSVs).
</ParamField>

<ParamField body="value_column" type="str | None">
  Column name containing expression values to read (e.g. `"tpm_unstranded"`).
</ParamField>

<ParamField body="strip_version" type="bool">
  When `True`, strips version suffixes from gene identifiers (e.g. `ENSG…00003.15` → `ENSG…00003`).
</ParamField>

<ResponseField name="returns" type="int">
  Number of records that were linked.
</ResponseField>

#### add\_labels

```python theme={null}
def add_labels(labels: dict[str, dict[str, Any]],
               *,
               by: str = "patient_id") -> None
```

Adds categorical labels to records.

<ParamField body="labels" type="dict[str, dict[str, Any]]" required>
  Mapping of `{identifier: {label_name: value}}`.
</ParamField>

<ParamField body="by" type="str">
  Key to match on — `"patient_id"` or `"wsi_id"`.
</ParamField>

#### add\_metadata

```python theme={null}
def add_metadata(metadata: dict[str, dict[str, Any]],
                 *,
                 by: str = "patient_id") -> None
```

Adds clinical/treatment metadata to records.

<ParamField body="metadata" type="dict[str, dict[str, Any]]" required>
  Mapping of `{identifier: {field: value}}`.
</ParamField>

<ParamField body="by" type="str">
  Key to match on — `"patient_id"` or `"wsi_id"`.
</ParamField>

#### add\_labels\_from\_csv

```python theme={null}
def add_labels_from_csv(path: str | Path, *, by: str = "patient_id") -> None
```

Adds labels and metadata from a CSV file.

The CSV must have a column matching *by* (default `patient_id`).  Columns prefixed with `label_` are treated as labels; remaining columns as metadata.

<ParamField body="path" type="str | Path" required>
  Path to the labels CSV.
</ParamField>

<ParamField body="by" type="str">
  Join key column name.
</ParamField>

#### num\_patients

```python theme={null}
@property
def num_patients() -> int
```

Number of unique patients in the cohort.

#### patients

```python theme={null}
@property
def patients() -> dict[str, PatientRecord]
```

Patient-indexed view of the cohort.

#### get\_patient

```python theme={null}
def get_patient(patient_id: str) -> PatientRecord
```

Returns the `PatientRecord` for a given patient.

<ParamField body="patient_id" type="str" required>
  Patient identifier string.
</ParamField>

<ResponseField name="returns" type="PatientRecord">
  The matching `PatientRecord`.
</ResponseField>

**Raises:**

* `KeyError` — If *patient\_id* is not found.

#### wsi\_ids

```python theme={null}
@property
def wsi_ids() -> list[str]
```

Ordered list of all WSI IDs.

#### patient\_ids

```python theme={null}
@property
def patient_ids() -> list[str]
```

Ordered list of unique patient IDs.

#### wsi\_paths

```python theme={null}
@property
def wsi_paths() -> list[Path | None]
```

Ordered list of WSI paths (may contain `None`).

#### pending

```python theme={null}
def pending(model_name: str, stage: str) -> list[WSIRecord]
```

Returns wsis that have not completed a stage for a model.

Checks whether the output file exists on disk.  Records with missing WSI paths are excluded.

<ParamField body="model_name" type="str" required>
  Model identifier.
</ParamField>

<ParamField body="stage" type="str" required>
  `"embed"` or `"predict"`.
</ParamField>

<ResponseField name="returns" type="list[WSIRecord]">
  List of `WSIRecord` entries still needing processing.
</ResponseField>

#### to\_csv

```python theme={null}
def to_csv(path: str | Path) -> Path
```

Writes input columns to a CSV file (labels + metadata).

For full persistence including outputs, use `save`.

<ParamField body="path" type="str | Path" required>
  Destination file path.
</ParamField>

<ResponseField name="returns" type="Path">
  The resolved output path.
</ResponseField>

#### summary

```python theme={null}
def summary() -> str
```

Returns a human-readable summary of the cohort.

<ResponseField name="returns" type="str">
  Multi-line summary string.
</ResponseField>