trx.trx_file_memmap#

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[source]#
trx.trx_file_memmap._get_dtype_little_endian(dtype: numpy.dtype | str | type) → numpy.dtype[source]#

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[source]#

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[source]#

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[source]#

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][source]#

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[source]#

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[source]#

Verify that filename extension is a valid datatype.

Parameters:
extstr

Filename extension.

Returns:
bool

True if the provided datatype is valid, False otherwise.

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[source]#

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][source]#

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][source]#

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][source]#

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][source]#

Remove empty TrxFiles from the list.

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

Get all dps and dpv keys from the TrxFile list.

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

Verify that space attributes are consistent across TrxFiles.

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

Verify dpv coherence across TrxFiles.

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

Verify dps coherence across TrxFiles.

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

Compute group length and dtype information.

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[source]#

Create a new TrxFile for concatenation.

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[source]#

Setup groups in the new TrxFile for concatenation.

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[source]#

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[source]#

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[source]#

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)[source]#

Core class of the TrxFile

header: dict[source]#
streamlines: Type[nibabel.streamlines.array_sequence.ArraySequence][source]#
groups: dict[source]#
data_per_streamline: dict[source]#
data_per_vertex: dict[source]#
data_per_group: dict[source]#
_copy_safe = True[source]#
__str__() → str[source]#

Generate the string for printing

__len__() → int[source]#

Define the length of the object

__getitem__(key) → Any[source]#

Slice all data in a consistent way

__deepcopy__() → Type[TrxFile][source]#
deepcopy() → Type[TrxFile][source]#

Create a deepcopy of the TrxFile.

Returns:
TrxFile

A deepcopied TrxFile of the current TrxFile.

_get_real_len() → Tuple[int, int][source]#

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][source]#

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][source]#

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][source]#

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[source]#

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()[source]#

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[source]#
_append_trx(trx: Type[TrxFile], extra_buffer: int = 0) → None[source]#

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][source]#

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][source]#

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][source]#

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)[source]#

Generate a valid TrxFile from a StatefulTractogram

static from_tractogram(tractogram, reference, dtype_dict=None)[source]#

Generate a valid TrxFile from a Nibabel Tractogram

to_tractogram(resize=False)[source]#

Convert a TrxFile to a nibabel Tractogram (in RAM)

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

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)[source]#

Convert a TrxFile to a valid StatefulTractogram (in RAM)

close() → None[source]#

Cleanup on-disk temporary folder and initialize an empty TrxFile