snake.core.engine.base

Engines are responsible for the acquisition of Kspace.

Module Contents

Classes

MetaEngine	MetaClass for engines.
BaseAcquisitionEngine	Base acquisition engine.

Data

AnyPath

API

snake.core.engine.base.AnyPath = None

class snake.core.engine.base.MetaEngine

Bases: snake._meta.MetaDCRegister

MetaClass for engines.

dunder_name: ClassVar[str] = 'engine'

class snake.core.engine.base.BaseAcquisitionEngine

Base acquisition engine.

Specific step can be overwritten in subclasses.

Parameters:

• model (str, optional) – The model to use, by default “simple”.

• snr (float, optional) – The signal to noise ratio, by default np.inf.

• slice_2d (bool, optional) – Whether to slice the 2D data, by default False.

__engine_name__: ClassVar[str] = None

__registry__: ClassVar[dict[str, type[snake.core.engine.base.BaseAcquisitionEngine]]] = None

log: ClassVar[logging.Logger] = None

model: str = 'simple'

snr: float = None

slice_2d: bool = False

_get_chunk_list(data_loader: snake.mrd_utils.MRDLoader) → collections.abc.Sequence[int]

abstract _job_trajectories(dataset: ismrmrd.Dataset, hdr: ismrmrd.xsd.ismrmrdHeader, sim_conf: snake.core.simulation.SimConfig, chunk: collections.abc.Sequence[int]) → numpy.typing.NDArray

static _job_get_T2s_decay(dwell_time_ms: float, echo_idx: int, n_samples: int, phantom: snake.core.phantom.Phantom) → numpy.typing.NDArray

abstract static _job_model_T2s(phantom: snake.core.phantom.Phantom, dyn_datas: list[snake.core.phantom.DynamicData], sim_conf: snake.core.simulation.SimConfig, trajectories: numpy.typing.NDArray, *args: Any, **kwargs: Any) → numpy.typing.NDArray

abstract static _job_model_simple(phantom: snake.core.phantom.Phantom, dyn_datas: list[snake.core.phantom.DynamicData], sim_conf: snake.core.simulation.SimConfig, trajectories: numpy.typing.NDArray, *args: Any, **kwargs: Any) → numpy.typing.NDArray

abstract _write_chunk_data(dataset: ismrmrd.Dataset, chunk: collections.abc.Sequence[int], chunk_data: numpy.typing.NDArray) → None

_acquire_ksp_job(filename: snake.core.engine.base.AnyPath, chunk: collections.abc.Sequence[int], tmp_dir: str, shared_phantom_props: tuple[str, snake.core.parallel.ArrayProps, snake.core.parallel.ArrayProps, snake.core.parallel.ArrayProps, snake.core.parallel.ArrayProps | None] | None = None, **kwargs: collections.abc.Mapping[str, Any]) → str

Entry point for worker.

This handles the io part (Read dataset, write partial k-space), and dispatch to specialized functions for getting the k-space.

__call__(filename: snake.core.engine.base.AnyPath, sampler: snake.core.sampling.BaseSampler, phantom: snake.core.phantom.Phantom, sim_conf: snake.core.simulation.SimConfig, handlers: list[snake.core.handlers.AbstractHandler] | snake.core.handlers.HandlerList | None = None, coil_cov: numpy.typing.NDArray | None = None, worker_chunk_size: int = 0, n_workers: int = 0, **kwargs: Any)

Perform the acquisition and fill the dataset.

Parameters:

• filename (AnyPath) – The path to the MRD file.

• sampler (BaseSampler) – The sampler to use.

• phantom (Phantom) – The phantom to use.

• sim_conf (SimConfig) – The simulation configuration.

• handlers (list[AbstractHandler] | HandlerList | None, optional) – The handlers to use, by default None.

• coil_cov (NDArray | None, optional) – The coil covariance matrix, by default None.

• worker_chunk_size (int, optional) – The size of the chunk to process for each worker, by default 0 (auto). Each worker w

• n_workers (int, optional) – The number of workers to use, by default 0 (auto). Half of CPU count will be used (This usually corresponds to the number of physical cores on the machine).

• kwargs (Any) – Additional keyword arguments, passed down to internal implementation.

Notes

This function is the main entry point for the acquisition engine. It will create the base dataset, and then dispatch the work to the workers.

Specific modeling steps are implemented in the _job_model_T2s and _job_model_simple methods.