Skip to content

cardiotensor

Modules:

analysis

Modules:

analysis_functions

Functions:

calculate_intensities 

calculate_intensities(img_helix: ndarray, start_point: tuple[int, int], end_point: tuple[int, int], angle_range: float = 5, N_line: int = 10, max_value: float | None = None, min_value: float | None = None) -> list[ndarray]

Calculate intensity profiles along multiple lines.

Parameters: img_helix (np.ndarray): The image array. start_point (Tuple[int, int]): The starting point of the line. end_point (Tuple[int, int]): The ending point of the line. angle_range (float, optional): The range of angles to consider in degrees. Default is 5. N_line (int, optional): The number of lines to generate. Default is 10. max_value (Optional[float], optional): Maximum value for intensity normalization. Default is None. min_value (Optional[float], optional): Minimum value for intensity normalization. Default is None.

Returns: List[np.ndarray]: List of intensity profiles for each line.

find_end_points 

find_end_points(start_point: tuple[float, float], end_point: tuple[float, float], angle_range: float, N_line: int) -> ndarray

Find the end points for lines at different angles within a range.

Parameters: start_point (Tuple[int, int]): The starting point of the main line. end_point (Tuple[int, int]): The ending point of the main line. angle_range (float): The range of angles to consider in degrees. N_line (int): The number of lines to generate within the range.

Returns: np.ndarray: Array of end points for the generated lines.

plot_intensity 

plot_intensity(intensity_profiles: list[ndarray], label_y: str = '', x_max_lim: float | None = None, x_min_lim: float | None = None, y_max_lim: float | None = None, y_min_lim: float | None = None) -> None

Plot intensity profiles with mean and percentile shading.

Parameters: intensity_profiles (List[np.ndarray]): List of intensity profiles. label_y (str, optional): Label for the y-axis. Default is an empty string. x_max_lim (Optional[float], optional): Maximum x-axis limit. Default is None. x_min_lim (Optional[float], optional): Minimum x-axis limit. Default is None. y_max_lim (Optional[float], optional): Maximum y-axis limit. Default is None. y_min_lim (Optional[float], optional): Minimum y-axis limit. Default is None.

save_intensity 

save_intensity(intensity_profiles: list[ndarray], save_path: str) -> None

Save intensity profiles to a CSV file.

Parameters: intensity_profiles (List[np.ndarray]): List of intensity profiles. save_path (str): Path to save the CSV file.

Returns: None

launcher

Modules:

slurm_launcher

Functions:

is_chunk_done 

is_chunk_done(output_dir: str, start: int, end: int, output_format: str = 'jp2') -> bool

Check if all output files (HA, IA, FA) for a given chunk are already present.

Parameters:

  • output_dir
    (str) –

    Base output directory containing HA/IA/FA folders.

  • start
    (int) –

    Start index of the chunk (inclusive).

  • end
    (int) –

    End index of the chunk (exclusive).

  • output_format
    (str, default: 'jp2' ) –

    File extension for the output images (e.g., "jp2", "tif").

Returns:

  • bool ( bool ) –

    True if all expected output files exist, False otherwise.

monitor_job_output 

monitor_job_output(output_directory: str, total_images: int, file_extension: str) -> None

Monitor the output directory until all images are processed.

Parameters:

Returns:

  • None –

    None

slurm_launcher 

slurm_launcher(conf_file_path: str) -> None

Launch Slurm jobs for 3D data processing.

Parameters:

Returns:

  • None –

    None

submit_job_to_slurm 

submit_job_to_slurm(executable_path: str, conf_file_path: str, start_image: int, end_image: int, N_chunk: int = 10, mem_needed: int = 64) -> int

Submit a Slurm job and return its job ID.

Parameters:

Returns:

  • int ( int ) –

    The Slurm job ID.

orientation

Modules:

orientation_computation_functions

Functions:

adjust_start_end_index 

adjust_start_end_index(start_index: int, end_index: int, N_img: int, padding_start: int = 0, padding_end: int = 0, is_test: bool = False, n_slice: int = 0) -> tuple[int, int]

Adjusts start and end indices for image processing, considering padding and test mode.

Parameters:

Returns:

  • tuple[int, int] –

    Tuple[int, int]: Adjusted start and end indices.

calculate_center_vector 

calculate_center_vector(points: ndarray) -> ndarray

Compute the linear regression vector for a given set of 3D points.

Parameters:

  • points
    (ndarray) –

    An Nx3 array of (x, y, z) coordinates representing the curved line.

Returns:

  • ndarray –

    np.ndarray: A single 3D unit vector representing the direction of the best-fit line.

calculate_structure_tensor 

calculate_structure_tensor(volume: ndarray, sigma: float, rho: float, truncate: float = 4.0, devices: list[str] | None = None, block_size: int = 200, use_gpu: bool = False, dtype: type = float32) -> tuple[ndarray, ndarray]

Calculates the structure tensor of a volume.

Parameters:

  • volume
    (ndarray) –

    The 3D volume data.

  • sigma
    (float) –

    sigma value for Gaussian smoothing.

  • rho
    (float) –

    rho value for Gaussian smoothing.

  • devices
    (Optional[list[str]], default: None ) –

    List of devices for parallel processing (e.g., ['cpu', 'cuda:0']).

  • block_size
    (int, default: 200 ) –

    Size of the blocks for processing. Default is 200.

  • use_gpu
    (bool, default: False ) –

    If True, uses GPU for calculations. Default is False.

Returns:

  • tuple[ndarray, ndarray] –

    tuple[np.ndarray, np.ndarray]: Eigenvalues and eigenvectors of the structure tensor.

compute_fraction_anisotropy 

compute_fraction_anisotropy(eigenvalues_2d: ndarray) -> ndarray

Computes Fractional Anisotropy (FA) from eigenvalues of a structure tensor.

Parameters:

Returns:

  • ndarray –

    np.ndarray: Fractional Anisotropy values.

compute_helix_and_transverse_angles 

compute_helix_and_transverse_angles(vector_field_2d: ndarray, center_point: tuple[int, int, int]) -> tuple[ndarray, ndarray]

Computes helix and transverse angles from a 2D vector field.

Parameters:

Returns:

  • tuple[ndarray, ndarray] –

    Tuple[np.ndarray, np.ndarray]: Helix and transverse angle arrays.

interpolate_points 

interpolate_points(points: list[tuple[float, float, float]], N_img: int) -> ndarray

Generates interpolated points using cubic spline interpolation for a given set of 3D points.

Parameters:

  • points
    (list[tuple[float, float, float]]) –

    A list of (x, y, z) points.

  • N_img
    (int) –

    The number of slices in the z-dimension.

Returns:

  • ndarray –

    np.ndarray: Array of interpolated points.

plot_images 

plot_images(img: ndarray, img_helix: ndarray, img_intrusion: ndarray, img_FA: ndarray, center_point: tuple[int, int, int], colormap_angle=None, colormap_FA=None) -> None

Plots images of the heart with helix, intrusion, and FA annotations.

Parameters:

Returns:

  • None –

    None

remove_padding 

remove_padding(volume: ndarray, val: ndarray, vec: ndarray, padding_start: int, padding_end: int) -> tuple[ndarray, ndarray, ndarray]

Removes padding from the volume, eigenvalues, and eigenvectors.

Parameters:

Returns:

  • tuple[ndarray, ndarray, ndarray] –

    Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: Adjusted data without padding.

rotate_vectors_to_new_axis 

rotate_vectors_to_new_axis(vector_field_slice: ndarray, new_axis_vec: ndarray) -> ndarray

Rotates a vector field slice to align with a new axis.

Parameters:

Returns:

  • ndarray –

    np.ndarray: Rotated vectors with the same shape as input.

write_images 

write_images(img_helix: ndarray, img_intrusion: ndarray, img_FA: ndarray, start_index: int, output_dir: str, output_format: str, output_type: str, z: int, colormap_angle=None, colormap_FA=None) -> None

Writes processed images to the specified directory.

Parameters:

Returns:

  • None –

    None

write_vector_field 

write_vector_field(vector_field_slice: ndarray, start_index: int, output_dir: str, slice_idx: int) -> None

Saves a vector field slice to the specified directory in .npy format.

Parameters:

Returns:

  • None –

    None

orientation_computation_pipeline

Functions:

check_already_processed 

check_already_processed(output_dir: str, start_index: int, end_index: int, write_vectors: bool, write_angles: bool, output_format: str) -> bool

Check whether all required output files already exist.

Parameters:

Returns:

  • bool ( bool ) –

    True if all expected files exist and are valid, else False.

compute_orientation 

compute_orientation(volume_path: str, mask_path: str | None = None, output_dir: str = './output', output_format: str = 'jp2', output_type: str = '8bit', sigma: float = 1.0, rho: float = 3.0, truncate: float = 4.0, axis_points: ndarray | None = None, vertical_padding: float | None = None, write_vectors: bool = False, write_angles: bool = True, use_gpu: bool = True, is_test: bool = False, n_slice_test: int | None = None, start_index: int = 0, end_index: int | None = None) -> None

Compute the orientation for a volume dataset.

Parameters:

  • volume_path
    (str) –

    Path to the 3D volume.

  • mask_path
    (str | None, default: None ) –

    Optional binary mask path.

  • output_dir
    (str, default: './output' ) –

    Output directory for results.

  • output_format
    (str, default: 'jp2' ) –

    Image format for results.

  • output_type
    (str, default: '8bit' ) –

    Image type ("8bit" or "rgb").

  • sigma
    (float, default: 1.0 ) –

    Noise scale for structure tensor.

  • rho
    (float, default: 3.0 ) –

    Integration scale for structure tensor.

  • truncate
    (float, default: 4.0 ) –

    Gaussian kernel truncation.

  • axis_points
    (ndarray | None, default: None ) –

    3D points defining LV axis for cylindrical coordinates.

  • vertical_padding
    (float | None, default: None ) –

    Padding slices for tensor computation.

  • write_vectors
    (bool, default: False ) –

    Whether to save eigenvectors.

  • write_angles
    (bool, default: True ) –

    Whether to save HA/IA/FA maps.

  • use_gpu
    (bool, default: True ) –

    Use GPU acceleration for tensor computation.

  • is_test
    (bool, default: False ) –

    If True, runs in test mode and outputs plots.

  • n_slice_test
    (int | None, default: None ) –

    Number of slices to process in test mode.

  • start_index
    (int, default: 0 ) –

    Start slice index.

  • end_index
    (int | None, default: None ) –

    End slice index (None = last slice).

compute_slice_angles_and_anisotropy 

compute_slice_angles_and_anisotropy(z: int, vector_field_slice: ndarray, img_slice: ndarray, center_point: ndarray, eigen_val_slice: ndarray, center_line: ndarray, output_dir: str, output_format: str, output_type: str, start_index: int, write_vectors: bool, write_angles: bool, is_test: bool) -> None

Compute helix angles, transverse angles, and fractional anisotropy for a slice.

Parameters:

Returns:

  • None –

    None

tractography

Modules:

generate_streamlines

Functions:

generate_streamlines_from_params 

generate_streamlines_from_params(vector_field_dir: str | Path, output_dir: str | Path, fa_dir: str | Path, ha_dir: str | Path, mask_path: str | Path | None = None, start_xyz: tuple[int, int, int] = (0, 0, 0), end_xyz: tuple[int | None, int | None, int | None] = (None, None, None), bin_factor: int = 1, num_seeds: int = 20000, fa_seed_min: float = 0.4, fa_threshold: float = 0.1, step_length: float = 0.5, max_steps: int | None = None, angle_threshold: float = 60.0, min_length_pts: int = 10, bidirectional: bool = True) -> None

Generate streamlines from a vector field and save to NPZ.

Parameters

vector_field_dir : str or Path Directory containing eigenvector volumes. output_dir : str or Path Directory where output NPZ will be saved. fa_dir : str or Path Directory containing FA volumes. ha_dir : str or Path Directory containing HA volumes. mask_path : str or Path, optional Path to mask volume to filter vectors. start_xyz, end_xyz : tuple[int] Cropping bounds in Z, Y, X (None = full dimension). bin_factor : int Spatial downsampling factor. num_seeds : int Number of random seed points for streamline generation. fa_threshold : float Minimum FA value to continue tracing. step_length : float Streamline step length in voxels. max_steps : int, optional Maximum number of steps per streamline. angle_threshold : float Maximum angle (degrees) allowed between steps. min_length_pts : int Minimum number of points for a valid streamline. bidirectional : bool Trace in both directions from each seed.

generate_streamlines_from_vector_field 

generate_streamlines_from_vector_field(vector_field: ndarray, seed_points: ndarray, fa_volume: ndarray | None = None, fa_threshold: float = 0.1, step_length: float = 0.5, max_steps: int | None = None, angle_threshold: float = 60.0, min_length_pts: int = 10, bidirectional: bool = True) -> list[list[tuple[float, float, float]]]

Given a 3D vector_field (shape = (3, Z, Y, X)) and a set of integer‐seed voxels, returns a list of streamlines (each streamline = a list of float (z,y,x) points), filtered so that only those longer than min_length_pts are kept. Optionally trace in both directions from each seed point.

trace_streamline 

trace_streamline(start_pt: tuple[float, float, float], vector_field: ndarray, fa_volume: ndarray | None = None, fa_threshold: float = 0.1, step_length: float = 0.5, max_steps: int | None = 1000, angle_threshold: float = 60.0, eps: float = 1e-10, direction: int = 1) -> list[tuple[float, float, float]]

Trace one streamline from start_pt (z,y,x) in the continuous vector_field. - Interpolate & normalize each sub‐step - Move forward by step_length voxels each step using RK4 - direction = +1 (default) or -1 to reverse integration direction

trilinear_interpolate_scalar 

trilinear_interpolate_scalar(volume: ndarray, pt: tuple[float, float, float]) -> float

Trilinearly interpolate a scalar volume at fractional point (z, y, x). Clamps to valid range.

trilinear_interpolate_vector 

trilinear_interpolate_vector(vector_field: ndarray, pt: tuple[float, float, float]) -> ndarray

Given a fractional (z,y,x), returns the trilinearly‐interpolated 3‐vector from vector_field (shape = (3, Z, Y, X)). Clamps to nearest voxel if out‐of‐bounds.

utils

Modules:

DataReader

Classes:

DataReader 

DataReader(path: str | Path)

Initializes the DataReader with a path to the volume.

Parameters:

  • path
    (str | Path) –

    Path to the volume directory or file.

Methods:

  • check_memory_requirement –

    Check if the dataset can fit in available memory.

  • load_volume –

    Loads the volume and resizes it to unbinned_shape if provided, using fast scipy.ndimage.zoom.

Attributes:

  • dtype (dtype) –

    Returns the data type of the volume.

  • shape (tuple[int, ...]) –

    Returns the shape of the volume as (Z, Y, X) or (Z, Y, X, C).

  • volume_size_gb (float) –

    Returns the total size of the volume in GB.

dtype property 
dtype: dtype

Returns the data type of the volume.

shape property 
shape: tuple[int, ...]

Returns the shape of the volume as (Z, Y, X) or (Z, Y, X, C).

volume_size_gb property 
volume_size_gb: float

Returns the total size of the volume in GB.

check_memory_requirement 
check_memory_requirement(shape, dtype, safety_factor=0.8)

Check if the dataset can fit in available memory.

Parameters:

  • shape
    (tuple[int]) –

    Shape of the array.

  • dtype
    (dtype) –

    NumPy dtype of the array.

  • safety_factor
    (float, default: 0.8 ) –

    Fraction of available memory allowed to be used.

load_volume 
load_volume(start_index: int = 0, end_index: int | None = None, unbinned_shape: tuple[int, int, int] | None = None) -> ndarray

Loads the volume and resizes it to unbinned_shape if provided, using fast scipy.ndimage.zoom.

Parameters:

  • start_index
    (int, default: 0 ) –

    Start index for slicing (for stacks).

  • end_index
    (int, default: None ) –

    End index for slicing (for stacks). If None, loads the entire stack.

  • unbinned_shape
    (tuple, default: None ) –

    Desired shape (Z, Y, X). If None, no resizing is done.

Returns:

  • ndarray –

    np.ndarray: Loaded volume.

downsampling

Functions:

chunked_downsample_vector_volume_mp 

chunked_downsample_vector_volume_mp(input_npy_dir: Path, bin_factor: int, output_dir: Path) -> None

Multi‐process + progress‐bar version of chunked_downsample_vector_volume. Reads a directory of per‐slice NumPy files (shape = (3, H, W) each), groups every bin_factor consecutive slices into blocks, averages, downsamples, renormalizes, and writes each block as one coarse slice in output_dir/bin{bin_factor}/eigen_vec/.

Parameters

input_npy_dir : Path Directory containing fine‐scale “eigen_vec_*.npy” files, each shape (3, H, W). bin_factor : int Number of fine Z‐slices per block. output_dir : Path Base output directory. Will create output_dir/bin{bin_factor}/eigen_vec/.

downsample_vector_volume 

downsample_vector_volume(input_npy: Path, bin_factor: int, output_dir: Path) -> None

Downsamples a vector volume using multiprocessing.

Parameters:

downsample_volume 

downsample_volume(input_path: Path, bin_factor: int, output_dir: Path, subfolder: str = 'HA', out_ext: str = 'tif', min_value: float = 0, max_value: float = 255) -> None

Downsamples a 3D image volume along the Z and XY axes and saves as 8-bit images.

This function reads a volumetric image dataset (e.g. TIFF stack) using DataReader, performs block averaging along the Z-axis and spatial downsampling in XY, then saves each resulting slice in a specified output directory as 8-bit images.

Parameters:

  • input_path
    (Path) –

    Path to the directory containing the image stack.

  • bin_factor
    (int) –

    Factor to downsample in XY and the number of Z-slices to average per output slice.

  • output_dir
    (Path) –

    Path to the output root directory.

  • subfolder
    (str, default: 'HA' ) –

    Subdirectory name under binX/ to place results (default: "HA").

  • out_ext
    (str, default: 'tif' ) –

    Output image format extension (e.g., 'tif', 'png').

  • min_value
    (float, default: 0 ) –

    Minimum value for intensity normalization to 8-bit.

  • max_value
    (float, default: 255 ) –

    Maximum value for intensity normalization to 8-bit.

Returns:

  • None –

    None

process_image_block 

process_image_block(file_list, block_idx, bin_factor, out_file, min_value, max_value)

Process a Z-block of images by averaging along the Z axis, downsampling in XY, converting to 8-bit, and writing to disk.

Parameters:

process_vector_block 

process_vector_block(block: list[Path], bin_factor: int, h: int, w: int, output_dir: Path, idx: int) -> None

Processes a single block of numpy files and saves the downsampled output.

Parameters:

  • block
    (List[Path]) –

    List of file paths to the numpy files in the block.

  • bin_factor
    (int) –

    Binning factor for downsampling.

  • h
    (int) –

    Height of the data block.

  • w
    (int) –

    Width of the data block.

  • output_dir
    (Path) –

    Path to the output directory.

  • idx
    (int) –

    Index of the current block.

utils

Functions:

convert_to_8bit 

convert_to_8bit(img: ndarray, perc_min: int = 0, perc_max: int = 100, min_value: float | None = None, max_value: float | None = None) -> ndarray

Converts a NumPy array to an 8-bit image.

Parameters:

  • img
    (ndarray) –

    Input image array.

  • perc_min
    (int, default: 0 ) –

    Minimum percentile for normalization. Default is 0.

  • perc_max
    (int, default: 100 ) –

    Maximum percentile for normalization. Default is 100.

  • min_value
    (Optional[float], default: None ) –

    Optional explicit minimum value.

  • max_value
    (Optional[float], default: None ) –

    Optional explicit maximum value.

Returns:

  • ndarray –

    np.ndarray: 8-bit converted image.

read_conf_file 

read_conf_file(file_path: str) -> dict[str, Any]

Reads and parses a configuration file into a dictionary.

Parameters:

Returns:

  • dict[str, Any] –

    Dict[str, Any]: Parsed configuration parameters.

Raises:

  • FileNotFoundError –

    If the configuration file does not exist.

  • ValueError –

    If expected numerical or array values are incorrectly formatted.