spectralbrain.runtime#

SpectralBrain runtime infrastructure.

Provides cross-cutting services consumed by all other modules: versioning, canonical type aliases, structured logging, Rich progress bars for various workloads, and Singularity/Apptainer container management for optional DL-based preprocessing.

This module has no intra-library imports so it can be imported first without circular dependencies.

Examples

>>> from spectralbrain.runtime import __version__, get_logger
>>> logger = get_logger("spectralbrain.core")
>>> logger.info("Loaded SpectralBrain %s", __version__)
>>> from spectralbrain.runtime import progress_simple
>>> with progress_simple("Computing HKS", total=20) as update:
...     for i, t in enumerate(t_values):
...         hks[:, i] = _hks_at_t(eigenvalues, eigenvectors, t)
...         update(1)

Module Attributes

VERSION_INFO

(major, minor, patch) as a comparable tuple.

Vertices

Vertex coordinates, shape (N, 3), float64, in mm (RAS).

Faces

Triangle face indices, shape (F, 3), int64, 0-indexed.

Points

Point-cloud coordinates, shape (N, 3), float64.

Normals

Unit normals, shape (N, 3), float64 — per-vertex or per-point.

Eigenvalues

LBO eigenvalue vector, shape (k,), float64, ascending.

Eigenvectors

LBO eigenvectors, shape (N, k), float64.

SparseMatrix

Any SciPy sparse matrix (CSR, CSC, COO).

MassMatrix

Lumped or consistent mass matrix, shape (N, N), sparse.

ScalarMap

Per-vertex scalar, shape (N,) — e.g. one HKS time-slice, Casorati curvature, or a Bayesian surprise score.

DescriptorMatrix

Multi-scale per-vertex descriptor, shape (N, T) — e.g. HKS evaluated at T time-scales, or WKS at T energies.

GlobalDescriptor

One vector per shape, shape (d,) — e.g. ShapeDNA, 3D Zernike moments, or a Fisher-vector aggregation.

DistanceMatrix

Pairwise distance/similarity matrix, shape (R, R) — e.g. ROI-to-ROI WESD in a geometric connectome.

LabelArray

Per-vertex / per-voxel integer label, shape (N,) — atlas ROI indices (Schaefer, aseg, etc.).

VolumeImage

NIfTI volume — typed as Any to avoid hard nibabel dep at import.

SurfaceImage

GIfTI surface — typed as Any to avoid hard nibabel dep at import.

ConnectomeMatrix

ROI × ROI spectral similarity/distance, shape (R, R).

NetworkMatrix

Network × Network summary, shape (K, K) — block-averaged :pydata:`ConnectomeMatrix` (e.g. 7×7 for Yeo networks).

PathLike

Anything pathlib.Path can consume.

T

Generic type variable used in container utilities.

Functions

get_logger([name, level, rich])

Return a configured logger for a SpectralBrain module.

progress_parallel([description, total])

Thread-safe progress bar for joblib.Parallel callbacks.

progress_simple([description, total])

Simple progress bar with ETA.

progress_spinner([description])

Indeterminate spinner for unknown-duration operations.

set_log_level(level)

Set the log level for the entire library.

Classes

AnalysisObjective(*values)

Objectives for recommend_descriptor().

AtlasScheme(*values)

Brain atlases supported by utils.atlas.

BackendName(*values)

Compute backends.

ContainerManager([cache_dir, registry])

Download, cache, verify, and execute Singularity containers.

ContainerSpec(name, sif_filename, ...[, ...])

Specification for one DL preprocessing container.

DescriptorType(*values)

Descriptor identifiers — used by recommend_descriptor() and the eligibility registry.

GeometryFormat(*values)

Geometry file formats recognised by io.loaders.

NestedProgress(outer_description, ...)

Two-level Rich progress for nested loops.

class spectralbrain.runtime.AnalysisObjective(*values)[source]#

Bases: Enum

Objectives for recommend_descriptor().

GROUP_DISCRIMINATION = 'group_discrimination'#
LATERALIZATION = 'lateralization'#
LONGITUDINAL_CHANGE = 'longitudinal_change'#
SUBREGION_DETECTION = 'subregion_detection'#
class spectralbrain.runtime.AtlasScheme(*values)[source]#

Bases: Enum

Brain atlases supported by utils.atlas.

AMYGDALA_NUCLEI = 'amygdala_nuclei'#
ASEG = 'aseg'#
BRAINNETOME = 'brainnetome'#
DESTRIEUX = 'destrieux'#
DKT = 'dkt'#
GLASSER_MMP = 'glasser_mmp'#
HIPPOCAMPAL_SUBFIELDS = 'hippocampal_subfields'#
JULICH_BRAIN = 'julich_brain'#
SCHAEFER_100 = 'schaefer100'#
SCHAEFER_1000 = 'schaefer1000'#
SCHAEFER_200 = 'schaefer200'#
SCHAEFER_400 = 'schaefer400'#
SCHAEFER_600 = 'schaefer600'#
SCHAEFER_800 = 'schaefer800'#
THALAMIC_NUCLEI = 'thalamic_nuclei'#
TIAN_S1 = 'tian_s1'#
TIAN_S2 = 'tian_s2'#
TIAN_S3 = 'tian_s3'#
TIAN_S4 = 'tian_s4'#
class spectralbrain.runtime.BackendName(*values)[source]#

Bases: Enum

Compute backends.

CUPY = 'cupy'#
JAX = 'jax'#
NUMPY = 'numpy'#
TORCH = 'torch'#
class spectralbrain.runtime.ContainerManager(cache_dir=PosixPath('/home/docs/.cache/spectralbrain/containers'), registry=None)[source]#

Bases: object

Download, cache, verify, and execute Singularity containers.

Containers are .sif files stored under a local cache directory (default ~/.cache/spectralbrain/containers/; override with the SPECTRALBRAIN_CACHE environment variable). Each is downloaded once on first use and verified via SHA-256.

Parameters:

Examples

>>> cm = ContainerManager()
>>> cm.status()
>>> cm.run("hdbet",
...        input_path="sub-01_T1w.nii.gz",
...        output_path="sub-01_brain.nii.gz")
clean(tool=None)[source]#

Remove cached container(s).

Parameters:

tool (str or None) – Specific tool, or None to clear all.

Return type:

None

ensure(tool)[source]#

Download tool container if not cached.

Parameters:

tool (str) – Registry key (e.g. "hdbet").

Returns:

Path – Local .sif path.

Raises:
Return type:

Path

is_cached(tool)[source]#

Check whether a container is already downloaded.

Parameters:

tool (str)

Return type:

bool

run(tool, *, input_path, output_path, extra_binds=None, extra_args=None, gpu=None)[source]#

Execute a containerised preprocessing tool.

Parameters:
  • tool (str) – Registry key.

  • input_path (PathLike) – Input file (bind-mounted read-only).

  • output_path (PathLike) – Output file (parent dir bind-mounted read-write).

  • extra_binds (list of str, optional) – Additional "host:container" bind specs.

  • extra_args (list of str, optional) – Arguments appended to the entrypoint.

  • gpu (bool or None) – Force GPU on/off; None = auto-detect.

Returns:

subprocess.CompletedProcess

Raises:
Return type:

CompletedProcess

status()[source]#

Print and return status of all registered containers.

Return type:

dict[str, dict[str, Any]]

property runtime_available: bool#

True if Singularity/Apptainer is installed.

property runtime_name: str#

'apptainer', 'singularity', or 'none'.

class spectralbrain.runtime.ContainerSpec(name, sif_filename, source_url, sha256, size_mb, entrypoint, gpu_required=True)[source]#

Bases: object

Specification for one DL preprocessing container.

Parameters:
  • name (str) – Human-readable tool name.

  • sif_filename (str) – Local cache filename.

  • source_url (str) – HTTPS download URL.

  • sha256 (str) – Expected SHA-256 digest for integrity check.

  • size_mb (int) – Approximate download size (shown to user).

  • entrypoint (str) – Command template — use {input} and {output} placeholders.

  • gpu_required (bool) – Needs --nv GPU passthrough.

entrypoint: str#
gpu_required: bool = True#
name: str#
sha256: str#
sif_filename: str#
size_mb: int#
source_url: str#
class spectralbrain.runtime.DescriptorType(*values)[source]#

Bases: Enum

Descriptor identifiers — used by recommend_descriptor() and the eligibility registry.

AGD = 'agd'#
ASMWD = 'asmwd'#
BATES_SP = 'bates_sp'#
BIHARMONIC = 'biharmonic'#
BKS = 'bks'#
CASORATI = 'casorati'#
COMMUTE_TIME = 'commute_time'#
DIFFUSION = 'diffusion'#
DWKS = 'dwks'#
ECCENTRICITY = 'eccentricity'#
ECT = 'ect'#
FINSLER_HKS = 'finsler_hks'#
FRACTAL_DIM = 'fractal_dim'#
GPS = 'gps'#
HKS = 'hks'#
IBKS = 'ibks'#
INTEGRAL_INVARIANT = 'integral_invariant'#
PHT = 'pht'#
SDF = 'sdf'#
SGW_HEAT = 'sgw_heat'#
SGW_MEXICAN_HAT = 'sgw_mexican_hat'#
SHAPEDNA = 'shapedna'#
SHAPE_INDEX = 'shape_index'#
SI_HKS = 'si_hks'#
WESD = 'wesd'#
WILLMORE_ENERGY = 'willmore_energy'#
WKS = 'wks'#
ZERNIKE_3D = 'zernike_3d'#
class spectralbrain.runtime.GeometryFormat(*values)[source]#

Bases: Enum

Geometry file formats recognised by io.loaders.

FREESURFER_ANNOT = 2#
FREESURFER_LABEL = 4#
FREESURFER_MORPH = 3#
FREESURFER_SURFACE = 1#
GIFTI_FUNC = 6#
GIFTI_LABEL = 7#
GIFTI_SURFACE = 5#
HDF5 = 14#
MGZ_VOLUME = 9#
NIFTI_VOLUME = 8#
NUMPY = 15#
OBJ = 11#
PLY = 10#
STL = 12#
VTK = 13#
spectralbrain.runtime.MassMatrix#

Lumped or consistent mass matrix, shape (N, N), sparse.

class spectralbrain.runtime.NestedProgress(outer_description, outer_total, inner_description, inner_total)[source]#

Bases: object

Two-level Rich progress for nested loops.

Parameters:
  • outer_description (str) – Outer-loop label (e.g. "Subjects").

  • outer_total (int) – Number of outer iterations.

  • inner_description (str) – Inner-loop label (e.g. "Structures").

  • inner_total (int) – Inner iterations per outer step.

Examples

>>> with NestedProgress("Subjects", 228, "ROIs", 44) as np:
...     for subj in subjects:
...         for roi in rois:
...             compute(subj, roi)
...             np.advance_inner()
...         np.advance_outer()
advance_inner(n=1)[source]#

Advance inner bar by n steps.

Parameters:

n (int)

Return type:

None

advance_outer(n=1)[source]#

Advance outer bar by n and reset the inner bar.

Parameters:

n (int)

Return type:

None

spectralbrain.runtime.SparseMatrix#

Any SciPy sparse matrix (CSR, CSC, COO).

The stiffness (Laplacian) matrix L and the mass matrix M live in this format. Backends convert to their native sparse types.

spectralbrain.runtime.SurfaceImage#

GIfTI surface — typed as Any to avoid hard nibabel dep at import.

spectralbrain.runtime.VolumeImage#

NIfTI volume — typed as Any to avoid hard nibabel dep at import.

spectralbrain.runtime.get_logger(name='spectralbrain', *, level=20, rich=True)[source]#

Return a configured logger for a SpectralBrain module.

Installs a rich.logging.RichHandler on first call (if Rich is available). Subsequent calls with the same name return the existing logger.

Parameters:
  • name (str) – Logger name — submodules should pass __name__.

  • level (int) – Logging level (default logging.INFO).

  • rich (bool) – Use Rich formatting when available.

Returns:

logging.Logger

Return type:

Logger

spectralbrain.runtime.progress_parallel(description='Parallel jobs', total=None)[source]#

Thread-safe progress bar for joblib.Parallel callbacks.

Parameters:
  • description (str) – Label.

  • total (int or None) – Total number of jobs.

Yields:

Callable[[int], None] – Thread-safe update(n).

Return type:

Generator[Callable[[int], None], None, None]

Examples

>>> from joblib import Parallel, delayed
>>> with progress_parallel("Subjects", total=228) as tick:
...     def _run(s):
...         result = process(s); tick(1); return result
...     Parallel(n_jobs=8)(delayed(_run)(s) for s in subjects)
spectralbrain.runtime.progress_simple(description='Processing', total=None)[source]#

Simple progress bar with ETA.

Parameters:
  • description (str) – Label shown left of the bar.

  • total (int or None) – Step count. None → indeterminate spinner.

Yields:

Callable[[int], None]update(n) advances the bar by n steps.

Return type:

Generator[Callable[[int], None], None, None]

Examples

>>> with progress_simple("Eigensolve", total=n_structures) as tick:
...     for s in structures:
...         decompose(s)
...         tick(1)
spectralbrain.runtime.progress_spinner(description='Working')[source]#

Indeterminate spinner for unknown-duration operations.

Parameters:

description (str) – Label next to the spinner.

Return type:

Generator[None, None, None]

Examples

>>> with progress_spinner("Downloading container"):
...     download_large_file(url, dest)
spectralbrain.runtime.set_log_level(level)[source]#

Set the log level for the entire library.

Parameters:

level (int or str) – E.g. logging.DEBUG or "WARNING".

Return type:

None

spectralbrain.runtime.ConnectomeMatrix#

ROI × ROI spectral similarity/distance, shape (R, R).

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.DescriptorMatrix#

Multi-scale per-vertex descriptor, shape (N, T) — e.g. HKS evaluated at T time-scales, or WKS at T energies.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.DistanceMatrix#

Pairwise distance/similarity matrix, shape (R, R) — e.g. ROI-to-ROI WESD in a geometric connectome.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.Eigenvalues#

LBO eigenvalue vector, shape (k,), float64, ascending.

Non-negative; λ₀ ≈ 0 (constant mode). λ₁ is the Fiedler value.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.Eigenvectors#

LBO eigenvectors, shape (N, k), float64.

Column i is eigenfunction φᵢ evaluated at each vertex/point. Orthonormal w.r.t. mass matrix: Φᵀ M Φ = I.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.Faces#

Triangle face indices, shape (F, 3), int64, 0-indexed.

alias of ndarray[tuple[Any, …], dtype[int64]]

spectralbrain.runtime.GlobalDescriptor#

One vector per shape, shape (d,) — e.g. ShapeDNA, 3D Zernike moments, or a Fisher-vector aggregation.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.LabelArray#

Per-vertex / per-voxel integer label, shape (N,) — atlas ROI indices (Schaefer, aseg, etc.).

alias of ndarray[tuple[Any, …], dtype[integer]]

spectralbrain.runtime.NetworkMatrix#

Network × Network summary, shape (K, K) — block-averaged :pydata:`ConnectomeMatrix` (e.g. 7×7 for Yeo networks).

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.Normals#

Unit normals, shape (N, 3), float64 — per-vertex or per-point.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.PathLike = str | os.PathLike#

Anything pathlib.Path can consume.

spectralbrain.runtime.Points#

Point-cloud coordinates, shape (N, 3), float64.

Semantically identical to :pydata:`Vertices` but used when there is no face connectivity (volumetric segmentation → voxel centroids).

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.ScalarMap#

Per-vertex scalar, shape (N,) — e.g. one HKS time-slice, Casorati curvature, or a Bayesian surprise score.

alias of ndarray[tuple[Any, …], dtype[floating]]

spectralbrain.runtime.VERSION_INFO: tuple[int, int, int] = (0, 1, 0)#

(major, minor, patch) as a comparable tuple.

spectralbrain.runtime.Vertices#

Vertex coordinates, shape (N, 3), float64, in mm (RAS).

alias of ndarray[tuple[Any, …], dtype[floating]]