Spec, RunPolicy, and YAML isomorphism

The Spec / DataSource / RunPolicy tripartition.

habit.spec holds algorithm specifications (what to compute), run policies (how to execute), the YAML isomorphism, and the legacy-config translation. Data location is deliberately absent: that is the DataSource contract’s concern.

User guide: Python API guide (v2.0) · Habitat Guide Habitat Guide. Chooser (habitat stages, parameter meanings, and allowed values): Habitat Spec component catalog. The live list of names and parameters come from each component constructor (and Registry.constructor_signature / Registry.create) — see Plugin introspection API (do not hand-copy a table).

Classes

`Spec`	Specification of ONE pluggable component.
`Stage`	One named step in a habitat dataflow.
`HabitatSpec`	Complete specification of a habitat analysis.
`MLSpec`	Complete specification of a tabular machine-learning analysis.
`RunPolicy`	Execution policy for a study run.
`LegacyConfigAdapter`	Translate frozen v0 YAML mappings into the v1 document model.
`LegacyTranslation`	Result of translating one v0 payload into the v1 document layout.
`MigrationReport`	Outcome of one `migrate_yaml` call.

Functions

`coerce_spec`	Coerce one component payload into a Spec.
`parse_feature_expression`	Parse one feature-tree expression into its canonical Spec tree.
`load_habitat_spec`	Load a `HabitatSpec` from its YAML form.
`save_habitat_spec`	Write a `HabitatSpec` as YAML.
`load_run_policy`	Load a `RunPolicy` from a YAML mapping.
`save_run_policy`	Write a `RunPolicy` as YAML.
`build_habitat_document`	Assemble a native v1 habitat document with expanded defaults.
`save_habitat_config`	Write a complete effective v1 habitat YAML for CLI / YAML-API replay.
`load_habitat_config`	Load and structurally validate a v1 habitat YAML document.
`detect_yaml_version`	Detect whether a YAML payload follows the v0 or v1 layout.
`migrate_yaml`	Migrate one v0 YAML config into the v1 document layout.
`validate_v1_document`	Validate the structure of a v1 document payload.

`Stage` and `HabitatSpec.stages` (source of truth)

A habitat analysis is an ordered list of named stages. Each Stage pairs a custom label with a Spec component. Stage names are labels, not role keywords: scientific roles are inferred from position + registry domain. Recommended labels (convention only): extract_voxel_features, preprocess1 / preprocess2 / …, partition, extract_supervoxel_features, pool, fit, assign, quantify. Leave role= unset for normal authoring (escape hatch only).

Strategy is inferred from the stage sequence:

Strategy	Stage-sequence signature	Notes
two_step	`partition` + `pool`	Post-pool feature preprocess is first-class
direct_pooling	`pool` only (no partition)	Post-pool feature preprocess is first-class
one_step	neither partition nor pool	Per-subject fit/assign; habitat ids not comparable across subjects

Partition without pool is rejected (validate_dataflow()).

Primary entry: fit_predict().

from habit.spec import HabitatSpec, Spec, Stage
import habit.recipes as recipes

# two_step shape (partition + pool)
two_step = HabitatSpec(
    name="demo_two_step",
    stages=(
        Stage("extract_voxel_features", Spec("raw", {"modalities": ["T1", "T2"]})),
        Stage("partition", Spec("slic", {"n_supervoxels": 50})),
        Stage("pool", Spec("pool")),
        Stage("fit", Spec("kmeans", {"n_habitats": 4})),
        Stage("assign", Spec("nearest_centroid")),
        Stage("quantify", Spec("volume")),
        Stage("quantify2", Spec("msi")),
        Stage("quantify3", Spec("ith_score")),
        Stage("quantify4", Spec("non_radiomics")),
        Stage("quantify5", Spec("graph")),
        # Heavy PyRadiomics families (opt-in; require pyradiomics):
        # Stage("quantify6", Spec("traditional")),
        # Stage("quantify7", Spec("whole_habitat")),
        # Stage("quantify8", Spec("each_habitat")),
    ),
    random_seed=42,
)

# direct_pooling shape (pool only; post-pool preprocess allowed)
direct = HabitatSpec(
    name="demo_direct",
    stages=(
        Stage("extract_voxel_features", Spec("raw", {"modalities": ["T1", "T2"]})),
        Stage("preprocess1", Spec("minmax", {"across_features": False})),
        Stage("pool", Spec("pool")),
        Stage(
            "preprocess2",
            Spec(
                "binning",
                {
                    "n_bins": 8,
                    "bin_strategy": "uniform",
                    "across_features": False,
                },
            ),
        ),
        Stage("fit", Spec("kmeans", {"n_habitats": 3})),
        Stage("assign", Spec("nearest_centroid")),
        Stage("quantify", Spec("volume")),
    ),
    random_seed=42,
)

# one_step shape (no partition, no pool)
one_step = HabitatSpec(
    name="demo_one_step",
    stages=(
        Stage("extract_voxel_features", Spec("raw", {"modalities": ["T1", "T2"]})),
        Stage("fit", Spec("kmeans", {"n_habitats": 3})),
        Stage("assign", Spec("nearest_centroid")),
        Stage("quantify", Spec("volume")),
    ),
    random_seed=42,
)

print(two_step.fingerprint())
print(two_step.describe_methods(style="radiology"))
# result = recipes.Study(spec=two_step).fit_predict(cohort)

Named-field constructor (deprecated)

The classic named fields (voxel_feature_extractor, supervoxelizer, habitat_model_fitter, *_preprocessors, …) and the pooling declaration remain a deprecated constructor through v3.x: they expand to the same internal stage list and keep historical fingerprints. HabitatSpec.from_dict still loads named-field YAML. New Python and new YAML should declare stages only. Removal is scheduled for v4.0.0.

Derived views: a pool stage ⇒ pooling="cohort" / definition_level="cohort"; otherwise "none" / "subject". definition_level is read-only.

Habitat factories two_step_habitat(), one_step_habitat(), and direct_pooling_habitat() return a Study whose design validates the shape their name promises before fit(). They remain the short path when you do not want to write the stage list by hand.

Save / load and runnable YAML

from habit.spec import (
    HabitatSpec,
    RunPolicy,
    Spec,
    Stage,
    load_habitat_spec,
    save_habitat_config,
    save_habitat_spec,
)

spec = HabitatSpec(
    name="two_step",
    stages=(
        Stage("extract_voxel_features", Spec("raw", {"modalities": ["T1", "T2"]})),
        Stage("partition", Spec("slic", {"n_supervoxels": 50})),
        Stage("pool", Spec("pool")),
        Stage("fit", Spec("kmeans", {"n_habitats": 4})),
        Stage("assign", Spec("nearest_centroid")),
        Stage("quantify", Spec("volume")),
    ),
    random_seed=42,
)

save_habitat_spec(spec, "habitat_spec.yaml")
restored = load_habitat_spec("habitat_spec.yaml")
payload = spec.to_dict()
again = HabitatSpec.from_dict(payload)

# Runnable v1 document (spec + data + policy + output, defaults expanded).
# Same file works with recipes.run_from_yaml and habit get-habitat --config.
save_habitat_config(
    "habitat_run.yaml",
    spec,
    data_source="demo_data/preprocessed",
    out_dir="out/habitat",
    policy=RunPolicy(workers=1, backend="serial", subject_timeout_sec=None),
)

Fingerprints: pure sugar forms (no explicit stages) keep the historical named-field + pooling payload for two_step / direct_pooling stability. Explicit stages records the ordered list (names + components) and random_seed.

Feature trees and the expression form

Copy-paste recipes (single-modality leaves, then combiners): Habitat Spec component catalog section 1. Nested trees: Feature composition.

Extraction stages accept a tree of nodes: leaves carry modality= / modalities= parameters, and combiner nodes nest their children under params["children"] as plain {"name", "params"} payloads. Any component entry may be written in two fingerprint-identical spellings — the structured mapping above, or the strict expression string parsed by parse_feature_expression() (coerce_spec() routes a string entry to the parser and a mapping entry to Spec.from_dict):

from habit.spec import HabitatSpec, Spec, Stage, parse_feature_expression

expr = parse_feature_expression(
    'concat(raw("T1"), local_entropy("T2", kernel_size=3))'
)
spec = HabitatSpec(
    name="tree",
    stages=(
        Stage("extract_voxel_features", expr),
        Stage("pool", Spec("pool")),
        Stage("fit", Spec("kmeans", {"n_habitats": 3})),
        Stage("assign", Spec("nearest_centroid")),
        Stage("quantify", Spec("volume")),
    ),
    random_seed=42,
)

# YAML dual form — a string entry is parsed the same way:
again = HabitatSpec.from_dict(
    {
        "name": "tree",
        "stages": [
            {
                "name": "extract_voxel_features",
                "component": 'concat(raw("T1"), local_entropy("T2", kernel_size=3))',
            },
            # ... remaining stages ...
        ],
    }
)

Expression grammar is deliberately strict: modality names are quoted strings, parameters are explicit key=value literals, children are nested calls (a quoted string among children becomes an implicit raw leaf). Bare v0.1-style identifiers are rejected with an explicit error rather than guessed — the legacy YAML adapter keeps its permissive parser for unquoted v0.1 expressions and only routes quoted expressions here, so old configs translate byte-identically while new configs get the tree.

`RunPolicy`

RunPolicy is the declarative snapshot of every scheduling concern. Field names match backend keyword arguments so the YAML policy: block and the Python form stay one-to-one.

Important

Windows + process pool: spawning workers re-imports your script. Put any call that starts ProcessPoolBackend (RunPolicy(backend="process"), workers > 1, or parallel_mode="isolated") inside:

if __name__ == "__main__":
    ...

Running the same code at module top level (or pasting it into a .py file without this guard) raises RuntimeError: ... bootstrapping phase on Windows. The habit CLI entry point is already safe; this applies to scripts / notebooks converted to scripts / pure-Python recipes. For a quick serial check, use RunPolicy(workers=1, backend="serial") (no spawn).

from habit.spec import RunPolicy, load_run_policy, save_run_policy

policy = RunPolicy(
    workers=4,
    backend="process",              # "serial" | "process"
    on_subject_failure="continue",  # or "fail_fast"
    subject_timeout_sec=900.0,
    parallel_mode="persistent",     # library default
    auto_retry_rounds=2,
)
save_run_policy(policy, "run_policy.yaml")
policy2 = load_run_policy("run_policy.yaml")

on_subject_failure="continue" isolates errors inside the execution backend. map() still raises ProcessingError by default; pass raise_on_failure=False (recipes / CLI) to proceed with successes — see Execution backends and Fault tolerance patterns.

Full field set (defaults from `habit/spec/policy.py`)

Field	Default	Role
`workers`	`1`	Parallel worker processes; `> 1` selects ProcessPool
`backend`	`"serial"`	`"serial"` or `"process"` (also forced when `workers > 1` or `parallel_mode=isolated`)
`subject_timeout_sec`	`900.0`	Per-subject wall-clock seconds; `None` disables (ProcessPool only; does not select backend)
`subject_spawn_timeout_sec`	`120.0`	Spawn-startup seconds; `None` disables (ProcessPool / isolated)
`graceful_shutdown_sec`	`15.0`	Seconds between `terminate()` and `kill()` on timeout
`on_subject_failure`	`"continue"`	`"continue"` or `"fail_fast"` (Serial + ProcessPool)
`oom_backoff`	`True`	Reduce workers after fatal `MemoryError` (ProcessPool only)
`oom_reduce_workers_by`	`1`	Workers subtracted per OOM step
`cap_workers_to_gpu_pool`	`False`	Clamp workers to the usable GPU pool
`resume`	`True`	Reuse checkpointed subject results when a store is attached
`checkpoint_dir`	`None`	Checkpoint root; resolved by CLI/recipe (not applied by `from_policy`)
`parallel_mode`	`"persistent"`	`"persistent"` or `"isolated"` (ProcessPool only)
`auto_retry_rounds`	`2`	In-run re-dispatch rounds for failed subjects; `0` disables
`retry_failed_subjects`	`False`	Re-queue checkpointed failures on the next resumed run
`force_rerun_subjects`	`()`	Subject IDs forced to recompute
`clear_checkpoint_on_success`	`False`	Remove the checkpoint directory after a clean run
`strict_checkpoint_hash`	`False`	Raise `CompatibilityError` on incompatible checkpoint fingerprint / legacy v0.1 layout (v0.1 parity)
`persistent_worker_max_consecutive_failures`	`1`	Restart a persistent worker after this many consecutive fatal failures
`persistent_worker_recycle_after_tasks`	`0`	Restart a persistent worker after this many successes (`0` disables)

CLI / run_from_yaml select ProcessPoolBackend when backend == "process", workers > 1, or parallel_mode == "isolated". A positive subject_timeout_sec alone does not force spawn. Details: Execution backends.

v0.1 YAML top-level keys vs `RunPolicy`

A v0.1 habitat document keeps parallel knobs at the YAML top level (processes, individual_subject_*, …). LegacyConfigAdapter renames them into the v1 policy: section. A native v1 document writes the right-hand names under policy: directly (see config/habitat/config_habitat_two_step_v1.yaml). Habitat field reference: Habitat Segmentation Configuration.

v0.1 top-level key	`RunPolicy` / v1 `policy:` field	Schema default	`RunPolicy` default
`processes`	`workers`	`2`	`1`
(implied by `processes > 1` )	`backend`	—	`"serial"` (set to `"process"` when translated `workers > 1`)
`individual_subject_timeout_sec`	`subject_timeout_sec`	`900.0`	`900.0`
`individual_subject_spawn_timeout_sec`	`subject_spawn_timeout_sec`	`120.0`	`120.0`
`individual_subject_graceful_shutdown_sec`	`graceful_shutdown_sec`	`15.0`	`15.0`
`on_subject_failure`	`on_subject_failure`	`"continue"`	`"continue"`
`oom_backoff`	`oom_backoff`	`True`	`True`
`oom_reduce_workers_by`	`oom_reduce_workers_by`	`1`	`1`
`cap_processes_to_gpu_pool`	`cap_workers_to_gpu_pool`	`False`	`False`
`resume`	`resume`	`True`	`True`
`checkpoint_dir`	`checkpoint_dir`	`None`	`None`
`individual_subject_parallel_mode`	`parallel_mode`	`"persistent"`	`"persistent"`
`individual_subject_auto_retry_rounds`	`auto_retry_rounds`	`2`	`2`
`retry_failed_subjects`	`retry_failed_subjects`	`False`	`False`
`force_rerun_subjects`	`force_rerun_subjects`	`[]`	`()`
`clear_checkpoint_on_success`	`clear_checkpoint_on_success`	`False`	`False`
`strict_checkpoint_hash`	`strict_checkpoint_hash`	`False`	`False`
`persistent_worker_max_consecutive_failures`	`persistent_worker_max_consecutive_failures`	`1`	`1`
`persistent_worker_recycle_after_tasks`	`persistent_worker_recycle_after_tasks`	`0`	`0`

Note the default gap on processes / workers: a bare v0.1 habitat YAML defaults to processes: 2 (process backend after translation), while a bare RunPolicy() defaults to workers=1, backend="serial". CLI / run_from_yaml select ProcessPool when workers > 1, backend="process", or parallel_mode="isolated" — not merely because the default subject_timeout_sec=900 is set.

Detect, validate, migrate YAML

from pathlib import Path
import yaml

from habit.spec import (
    LegacyConfigAdapter,
    detect_yaml_version,
    migrate_yaml,
    validate_v1_document,
)

payload = yaml.safe_load(
    Path("config/habitat/config_habitat_two_step.yaml").read_text(
        encoding="utf-8"
    )
)
version = detect_yaml_version(payload)  # "v0" | "v1"

# Structural validation of a v1 document
# validate_v1_document(v1_payload, workflow="habitat")

# Migrate v0 -> v1 (dry-run)
report = migrate_yaml(
    "config/habitat/config_habitat_two_step.yaml",
    dry_run=True,
    workflow="habitat",
)
print(report.diff)
print(report.document)

# Lower-level translation
translation = LegacyConfigAdapter().translate(payload, "habitat")
# translation.document["spec"] / translation.document["policy"]

Run the translated spec with a recipe

v0.1 YAML selects the habitat design via habitat_segmentation.clustering_mode. Translation turns that knob into named-field sugar plus a derived pooling declaration (one_step → "none"; two_step / direct_pooling → "cohort"). Native v1 documents may also declare explicit stages. Either way, fit_predict() runs the shared stage executor. Mode-named aliases remain as thin validators:

`clustering_mode` (YAML)	Inferred stage signature / sugar	Alias (all dispatch to `Study.fit_predict`)
`two_step`	partition + pool (sugar: `pooling="cohort"` + supervoxelizer)	`two_step_habitat()`
`one_step`	neither (sugar: `pooling="none"`)	`one_step_habitat()`
`direct_pooling`	pool only (sugar: `pooling="cohort"`, no supervoxelizer)	`direct_pooling_habitat()`

Pattern: load the YAML, translate with LegacyConfigAdapter, build a HabitatSpec, then call fit_predict():

from pathlib import Path

import yaml
from habit.datasets import make_synthetic_cohort
from habit.spec import HabitatSpec, LegacyConfigAdapter
import habit.recipes as recipes

payload = yaml.safe_load(
    Path("config/habitat/config_habitat_two_step.yaml").read_text(
        encoding="utf-8"
    )
)
translation = LegacyConfigAdapter().translate(payload, "habitat")
spec = HabitatSpec.from_dict(translation.document["spec"])

# Modalities must match the spec's feature expression.
cohort = make_synthetic_cohort(
    n_subjects=4,
    modalities=("pre_contrast", "LAP", "PVP", "delay_3min"),
    rng=42,
)
# cohort = DirectoryDataSource(...).load()  # real data on disk

result = recipes.Study(spec=spec).fit_predict(cohort)
result.save("out/study")

Workflow aliases accepted by migrate / validate / adapter:

preprocess, habitat, extract, radiomics, model, cv, compare, icc, retest, sort-dicom.

CLI

habit check-config -c PATH — auto-detects v0/v1
habit migrate-config -c PATH — writes a v1 document

See Command reference.