trx.trx_file_memmap

Core TrxFile class with memory-mapped data access.

Attributes

dipy_available

Classes

TrxFile

Core class of the TrxFile.

Functions

_get_dtype_little_endian(→ numpy.dtype)	Convert a dtype to its little-endian equivalent.
_ensure_little_endian(→ numpy.ndarray)	Ensure array data is in little-endian byte order for writing.
_append_last_offsets(→ numpy.ndarray)	Append the last element of offsets from header information.
_generate_filename_from_data(→ str)	Determine the data type from array data and generate the appropriate filename.
_split_ext_with_dimensionality(→ Tuple[str, int, str])	Take a filename and split it into its components.
_compute_lengths(→ numpy.ndarray)	Compute lengths from offsets.
_is_dtype_valid(→ bool)	Verify that filename extension is a valid datatype.
_dichotomic_search(→ int)	Find where data of a contiguous array is actually ending.
_create_memmap(, dtype, offset, order)	Wrap memmap creation to support empty arrays.
load(→ Type[TrxFile])	Load a TrxFile (compressed or not).
load_from_zip(→ Type[TrxFile])	Load a TrxFile from a single zipfile.
load_from_directory(→ Type[TrxFile])	Load a TrxFile from a folder containing memmaps.
_filter_empty_trx_files(→ List[TrxFile])	Remove empty TrxFiles from the list.
_get_all_data_keys(→ Tuple[set, set])	Get all dps and dpv keys from the TrxFile list.
_check_space_attributes(→ None)	Verify that space attributes are consistent across TrxFiles.
_verify_dpv_coherence(→ None)	Verify dpv coherence across TrxFiles.
_verify_dps_coherence(→ None)	Verify dps coherence across TrxFiles.
_compute_groups_info(→ Tuple[dict, dict])	Compute group length and dtype information.
_create_new_trx_for_concatenation(→ TrxFile)	Create a new TrxFile for concatenation.
_setup_groups_for_concatenation(→ None)	Setup groups in the new TrxFile for concatenation.
concatenate(→ TrxFile)	Concatenate multiple TrxFile together, with support for preallocation.
save(→ None)	Save a TrxFile (compressed or not).
zip_from_folder(→ None)	Zip on-disk memmaps into a single file.

Module Contents

trx.trx_file_memmap.dipy_available = True

trx.trx_file_memmap._get_dtype_little_endian(dtype: numpy.dtype | str | type) → numpy.dtype

Convert a dtype to its little-endian equivalent.

The TRX file format uses little-endian byte order for cross-platform compatibility. This function ensures that dtypes are always interpreted as little-endian when reading/writing TRX files.

Parameters:

dtypenp.dtype, str, or type

Input dtype specification (e.g., np.float32, ‘float32’, ‘>f4’).

Returns:

np.dtype

Little-endian dtype. For single-byte types (uint8, int8, bool), returns the original dtype as endianness is not applicable.

trx.trx_file_memmap._ensure_little_endian(arr: numpy.ndarray) → numpy.ndarray

Ensure array data is in little-endian byte order for writing.

Parameters:

arrnp.ndarray

Input array.

Returns:

np.ndarray

Array with little-endian byte order. Returns a copy if conversion was needed, otherwise returns the original array.

trx.trx_file_memmap._append_last_offsets(nib_offsets: numpy.ndarray, nb_vertices: int) → numpy.ndarray

Append the last element of offsets from header information.

Parameters:

nib_offsetsnp.ndarray

Array of offsets with the last element being the start of the last streamline (nibabel convention).

nb_verticesint

Total number of vertices in the streamlines.

Returns:

np.ndarray

Offsets array (VTK convention).

trx.trx_file_memmap._generate_filename_from_data(arr: numpy.ndarray, filename: str) → str

Determine the data type from array data and generate the appropriate filename.

Parameters:

arrnp.ndarray

A NumPy array (1-2D, otherwise ValueError raised).

filenamestr

The original filename.

Returns:

str

An updated filename with appropriate extension.

trx.trx_file_memmap._split_ext_with_dimensionality(filename: str) → Tuple[str, int, str]

Take a filename and split it into its components.

Parameters:

filenamestr

Input filename.

Returns:

tuple

A tuple of (basename, dimension, extension).

trx.trx_file_memmap._compute_lengths(offsets: numpy.ndarray) → numpy.ndarray

Compute lengths from offsets.

Parameters:

offsetsnp.ndarray

An array of offsets.

Returns:

np.ndarray

An array of lengths.

trx.trx_file_memmap._is_dtype_valid(ext: str) → bool

Verify that filename extension is a valid datatype.

Parameters:

extstr

Filename extension.

Returns:

bool

True if the provided datatype is valid, False otherwise.

trx.trx_file_memmap._dichotomic_search(x: numpy.ndarray, l_bound: int | None = None, r_bound: int | None = None) → int

Find where data of a contiguous array is actually ending.

Parameters:

xnp.ndarray

Array of values.

l_boundint, optional

Lower bound index for search.

r_boundint, optional

Upper bound index for search.

Returns:

int

Index at which array value is 0 (if possible), otherwise returns -1.

trx.trx_file_memmap._create_memmap(filename: str, mode: str = 'r', shape: Tuple = (1,), dtype: numpy.dtype = np.float32, offset: int = 0, order: str = 'C') → numpy.ndarray

Wrap memmap creation to support empty arrays.

Parameters:

filenamestr

Filename where the empty memmap should be created.

modestr, optional

File open mode (see np.memmap for options). Default is ‘r’.

shapetuple, optional

Shape of memmapped array. Default is (1,).

dtypenp.dtype, optional

Datatype of memmapped array. Default is np.float32.

offsetint, optional

Offset of the data within the file. Default is 0.

orderstr, optional

Data representation on disk (‘C’ or ‘F’). Default is ‘C’.

Returns:

np.ndarray

Memory-mapped array or a zero-filled array if shape[0] is 0.

trx.trx_file_memmap.load(input_obj: str, check_dpg: bool = True) → Type[TrxFile]

Load a TrxFile (compressed or not).

Parameters:

input_objstr

A directory name or filepath to the TRX data.

check_dpgbool, optional

Whether to check group metadata. Default is True.

Returns:

TrxFile

TrxFile object representing the read data.

trx.trx_file_memmap.load_from_zip(filename: str) → Type[TrxFile]

Load a TrxFile from a single zipfile.

Note: Does not work with compressed zipfiles.

Parameters:

filenamestr

Path of the zipped TrxFile.

Returns:

TrxFile

TrxFile representing the read data.

trx.trx_file_memmap.load_from_directory(directory: str) → Type[TrxFile]

Load a TrxFile from a folder containing memmaps.

Parameters:

directorystr

Path of the directory containing TRX data.

Returns:

TrxFile

TrxFile representing the read data.

trx.trx_file_memmap._filter_empty_trx_files(trx_list: List[TrxFile]) → List[TrxFile]

Remove empty TrxFiles from the list.

Parameters:

trx_listlist of TrxFile class instances

Collection of tractograms to filter.

Returns:

list of TrxFile class instances

Only entries containing at least one streamline.

trx.trx_file_memmap._get_all_data_keys(trx_list: List[TrxFile]) → Tuple[set, set]

Get all dps and dpv keys from the TrxFile list.

Parameters:

trx_listlist of TrxFile class instances

Collection of tractograms.

Returns:

tuple of set

Sets of data_per_streamline keys and data_per_vertex keys.

trx.trx_file_memmap._check_space_attributes(trx_list: List[TrxFile]) → None

Verify that space attributes are consistent across TrxFiles.

Parameters:

trx_listlist of TrxFile

Tractograms to compare for affine and dimension consistency.

Raises:

ValueError

If voxel-to-RASMM matrices or dimensions differ.

trx.trx_file_memmap._verify_dpv_coherence(trx_list: List[TrxFile], all_dpv: set, ref_trx: TrxFile, delete_dpv: bool) → None

Verify dpv coherence across TrxFiles.

Parameters:

trx_listlist of TrxFile class instances

Tractograms being concatenated.

all_dpvset

Union of data_per_vertex keys across tractograms.

ref_trxTrxFile class instance

Reference tractogram for dtype/key checks.

delete_dpvbool

Drop mismatched dpv keys instead of raising when True.

Raises:

ValueError

If dpv keys or dtypes differ and delete_dpv is False.

trx.trx_file_memmap._verify_dps_coherence(trx_list: List[TrxFile], all_dps: set, ref_trx: TrxFile, delete_dps: bool) → None

Verify dps coherence across TrxFiles.

Parameters:

trx_listlist of TrxFile class instances

Tractograms being concatenated.

all_dpsset

Union of data_per_streamline keys across tractograms.

ref_trxTrxFile class instance

Reference tractogram for dtype/key checks.

delete_dpsbool

Drop mismatched dps keys instead of raising when True.

Raises:

ValueError

If dps keys or dtypes differ and delete_dps is False.

trx.trx_file_memmap._compute_groups_info(trx_list: List[TrxFile]) → Tuple[dict, dict]

Compute group length and dtype information.

Parameters:

trx_listlist of TrxFile class instances

Tractograms being concatenated.

Returns:

tuple of dict

(group lengths, group dtypes) keyed by group name.

trx.trx_file_memmap._create_new_trx_for_concatenation(trx_list: List[TrxFile], ref_trx: TrxFile, delete_dps: bool, delete_dpv: bool, delete_groups: bool) → TrxFile

Create a new TrxFile for concatenation.

Parameters:

trx_listlist of TrxFile class instances

Input tractograms to concatenate.

ref_trxTrxFile class instance

Reference tractogram for header/dtype template.

delete_dpsbool

Drop data_per_streamline keys not shared.

delete_dpvbool

Drop data_per_vertex keys not shared.

delete_groupsbool

Drop groups when metadata differ.

Returns:

TrxFile

Empty TRX ready to receive concatenated data.

trx.trx_file_memmap._setup_groups_for_concatenation(new_trx: TrxFile, trx_list: List[TrxFile], all_groups_len: dict, all_groups_dtype: dict, delete_groups: bool) → None

Setup groups in the new TrxFile for concatenation.

Parameters:

new_trxTrxFile class instance

Destination tractogram.

trx_listlist of TrxFile class instances

Source tractograms.

all_groups_lendict

Mapping of group name to total length.

all_groups_dtypedict

Mapping of group name to dtype.

delete_groupsbool

If True, skip creating group arrays.

trx.trx_file_memmap.concatenate(trx_list: List[TrxFile], delete_dpv: bool = False, delete_dps: bool = False, delete_groups: bool = False, check_space_attributes: bool = True, preallocation: bool = False) → TrxFile

Concatenate multiple TrxFile together, with support for preallocation.

Parameters:

trx_listlist of TrxFile

A list containing TrxFiles to concatenate.

delete_dpvbool, optional

Delete dpv keys that do not exist in all the provided TrxFiles. Default is False.

delete_dpsbool, optional

Delete dps keys that do not exist in all the provided TrxFiles. Default is False.

delete_groupsbool, optional

Delete all the groups that currently exist in the TrxFiles. Default is False.

check_space_attributesbool, optional

Verify that dimensions and size of data are similar between all the TrxFiles. Default is True.

preallocationbool, optional

Preallocated TrxFile has already been generated and is the first element in trx_list. Note: delete_groups must be set to True as well. Default is False.

Returns:

TrxFile

TrxFile representing the concatenated data.

trx.trx_file_memmap.save(trx: TrxFile, filename: str, compression_standard: Any = zipfile.ZIP_STORED) → None

Save a TrxFile (compressed or not).

Parameters:

trxTrxFile

The TrxFile to save.

filenamestr

The path to save the TrxFile to.

compression_standardint, optional

The compression standard to use, as defined by the ZipFile library. Default is zipfile.ZIP_STORED.

trx.trx_file_memmap.zip_from_folder(directory: str, filename: str, compression_standard: Any = zipfile.ZIP_STORED) → None

Zip on-disk memmaps into a single file.

Parameters:

directorystr

The path to the on-disk memmap directory.

filenamestr

The path where the zip file should be created.

compression_standardint, optional

The compression standard to use, as defined by the ZipFile library. Default is zipfile.ZIP_STORED.

class trx.trx_file_memmap.TrxFile(nb_vertices: int | None = None, nb_streamlines: int | None = None, init_as: Type[TrxFile] | None = None, reference: str | dict | Type[nibabel.nifti1.Nifti1Image] | Type[nibabel.streamlines.trk.TrkFile] | Type[nibabel.nifti1.Nifti1Header] | None = None)

Core class of the TrxFile.

Parameters:

nb_verticesint, optional

The number of vertices to use in the new TrxFile.

nb_streamlinesint, optional

The number of streamlines in the new TrxFile.

init_asTrxFile class instance, optional

A TrxFile to use as reference.

referencestr, dict, Nifti1Image, TrkFile, or Nifti1Header, optional

A Nifti or Trk file/obj to use as reference.

header: dict

streamlines: Type[nibabel.streamlines.array_sequence.ArraySequence]

groups: dict

data_per_streamline: dict

data_per_vertex: dict

data_per_group: dict

_copy_safe = True

__str__() → str

Generate the string for printing

__len__() → int

Define the length of the object

__getitem__(key) → Any

Slice all data in a consistent way

__deepcopy__() → Type[TrxFile]

Return a deep copy of the TrxFile.

Parameters:

self

TrxFile class instance.

Returns:

TrxFile class instance

Deep-copied instance.

deepcopy() → Type[TrxFile]

Create a deepcopy of the TrxFile.

Returns:

TrxFile

A deepcopied TrxFile of the current TrxFile.

_get_real_len() → Tuple[int, int]

Get the real size of data (ignoring zeros of preallocation).

Returns:

tuple of int

A tuple (strs_end, pts_end) representing the index of the last streamline and the total length of all the streamlines.

_copy_fixed_arrays_from(trx: Type[TrxFile], strs_start: int = 0, pts_start: int = 0, nb_strs_to_copy: int | None = None) → Tuple[int, int]

Fill a TrxFile using another and start indexes (preallocation).

Parameters:

trxTrxFile

TrxFile to copy data from.

strs_startint, optional

The start index of the streamline. Default is 0.

pts_startint, optional

The start index of the point. Default is 0.

nb_strs_to_copyint, optional

The number of streamlines to copy. If not set, will copy all.

Returns:

tuple of int

A tuple (strs_end, pts_end) representing the end of the copied streamlines and end of copied points.

static _initialize_empty_trx(nb_streamlines: int, nb_vertices: int, init_as: Type[TrxFile] | None = None) → Type[TrxFile]

Create on-disk memmaps of a certain size (preallocation).

Parameters:

nb_streamlinesint

The number of streamlines that the empty TrxFile will be initialized with.

nb_verticesint

The number of vertices that the empty TrxFile will be initialized with.

init_asTrxFile, optional

A TrxFile to initialize the empty TrxFile with.

Returns:

TrxFile

An empty TrxFile preallocated with a certain size.

_create_trx_from_pointer(dict_pointer_size: dict, root_zip: str | None = None, root: str | None = None) → Type[TrxFile]

Create a TrxFile after reading the structure of a zip/folder.

Parameters:

headerdict

A TrxFile header dictionary which will be used for the new TrxFile.

dict_pointer_sizedict

A dictionary containing the filenames of all the files within the TrxFile disk file/folder.

root_zipstr, optional

The path of the ZipFile pointer.

rootstr, optional

The dirname of the ZipFile pointer.

Returns:

TrxFile

A TrxFile constructed from the pointer provided.

resize(nb_streamlines: int | None = None, nb_vertices: int | None = None, delete_dpg: bool = False) → None

Remove the unused portion of preallocated memmaps.

Parameters:

nb_streamlinesint, optional

The number of streamlines to keep.

nb_verticesint, optional

The number of vertices to keep.

delete_dpgbool, optional

Remove data_per_group when resizing. Default is False.

get_dtype_dict()

Get the dtype dictionary for the TrxFile.

Returns:

dict

A dictionary containing the dtype for each data element.

append(obj, extra_buffer: int = 0) → None

Append another tractogram-like object to this TRX.

Parameters:

objTrxFile or Tractogram or StatefulTractogram class instance

Object whose streamlines and associated data will be appended.

extra_bufferint, optional

Additional preallocation buffer for streamlines (in count).

Returns:

None

Mutates the current TrxFile in-place.

_append_trx(trx: Type[TrxFile], extra_buffer: int = 0) → None

Append a TrxFile to another (with buffer support).

Parameters:

trxTrxFile

The TrxFile to append to the current TrxFile.

extra_bufferint, optional

The additional buffer space required to append data. Default is 0.

get_group(key: str, keep_group: bool = True, copy_safe: bool = False) → Type[TrxFile]

Get a particular group from the TrxFile.

Parameters:

keystr

The group name to select.

keep_groupbool, optional

Make sure group exists in returned TrxFile. Default is True.

copy_safebool, optional

Perform a deepcopy. Default is False.

Returns:

TrxFile

A TrxFile exclusively containing data from said group.

select(indices: numpy.ndarray, keep_group: bool = True, copy_safe: bool = False) → Type[TrxFile]

Get a subset of items, always pointing to the same memmaps.

Parameters:

indicesnp.ndarray

The list of indices of elements to return.

keep_groupbool, optional

Ensure group is returned in output TrxFile. Default is True.

copy_safebool, optional

Perform a deep-copy. Default is False.

Returns:

TrxFile

A TrxFile containing data originating from the selected indices.

static from_lazy_tractogram(obj: [nibabel.streamlines.tractogram.LazyTractogram], reference, extra_buffer: int = 0, chunk_size: int = 10000, dtype_dict: dict = None) → Type[TrxFile]

Create a TrxFile from a LazyTractogram with buffer support.

Parameters:

objLazyTractogram

The LazyTractogram to convert.

referenceobject

Reference for spatial information.

extra_bufferint, optional

The buffer space between reallocation. This number should be a number of streamlines. Use 0 for no buffer. Default is 0.

chunk_sizeint, optional

The number of streamlines to save at a time. Default is 10000.

dtype_dictdict, optional

Dictionary specifying dtypes for positions, offsets, dpv, and dps.

Returns:

TrxFile

A TrxFile created from the LazyTractogram.

static from_sft(sft, dtype_dict=None)

Generate a TrxFile from a StatefulTractogram.

Parameters:

sftStatefulTractogram class instance

Input tractogram.

dtype_dictdict or None, optional

Mapping of target dtypes for positions, offsets, dpv, and dps. When None, uses sft.dtype_dict or sensible defaults.

Returns:

TrxFile

TRX representation of the StatefulTractogram.

static from_tractogram(tractogram, reference, dtype_dict=None)

Generate a TrxFile from a nibabel Tractogram.

Parameters:

tractogramnibabel.streamlines.Tractogram class instance

Input tractogram to convert.

referenceobject

Reference anatomy used to populate header fields.

dtype_dictdict or None, optional

Mapping of target dtypes for positions, offsets, dpv, and dps.

Returns:

TrxFile class instance

TRX representation of the tractogram.

to_tractogram(resize=False)

Convert this TrxFile to a nibabel Tractogram.

Parameters:

resizebool, optional

If True, resize to actual data length before conversion.

Returns:

nibabel.streamlines.Tractogram class instance

Tractogram containing streamlines and metadata.

to_memory(resize: bool = False) → Type[TrxFile]

Convert a TrxFile to a RAM representation.

Parameters:

resizebool, optional

Resize TrxFile when converting to RAM representation. Default is False.

Returns:

TrxFile

A non memory-mapped TrxFile.

to_sft(resize=False)

Convert this TrxFile to a StatefulTractogram.

Parameters:

resizebool, optional

If True, resize to actual data length before conversion.

Returns:

StatefulTractogram class instance or None

StatefulTractogram object, or None if dipy is unavailable.

close() → None

Cleanup on-disk temporary folder and memmaps.

Returns:

None

Releases file handles and removes temporary storage.