oxytcmri.domain.use_cases.compute_lesions_volumes

Classes:

Name	Description
BrainLesionsVolume	Represents the volume of brain lesions for a specific MRI exam, DTI metric,
BrainLesionsVolumeRepository	Abstract base class for Brain Lesions Volume repository.
ComputeBrainLesionsVolumes	Use-case for computing the volumes of brain lesions in MRI exams.

Attributes:

Name	Type	Description
logger

logger = getLogger(__name__) module-attribute

BrainLesionsVolume(mri_exam_id, dti_metric, region_of_interest, abnormal_value_type, value_ml) dataclass

Represents the volume of brain lesions for a specific MRI exam, DTI metric, and region of interest.

Attributes:

Name	Type	Description
mri_exam_id	MRIExamId	The ID of the MRI exam.
dti_metric	DTIMetric	The DTI metric for which the volume is computed.
region_of_interest	Optional[RegionOfInterest]	The region of interest for which the volume is computed.
abnormal_value_type	AbnormalValueType	The type of abnormal value (HIGH or LOW) for which the volume is computed.
value_ml	float	The computed volume in milliliters (ml) of the brain lesions.

mri_exam_id instance-attribute

dti_metric instance-attribute

region_of_interest instance-attribute

abnormal_value_type instance-attribute

value_ml instance-attribute

BrainLesionsVolumeRepository

Bases: Repository[BrainLesionsVolume, None], ABC

Abstract base class for Brain Lesions Volume repository. Defines the interface for retrieving brain lesions volume data.

ComputeBrainLesionsVolumes(repositories_registry, dispatcher=None, overwrite_database=False)

Use-case for computing the volumes of brain lesions in MRI exams.

Initializes the SegmentDtiAbnormalValues use-case.

Methods:

Name	Description
compute_all_brain_lesions	Computes the brain lesions for all patients in the SubjectRepository.
compute_brain_lesions_associated_to_mri_exam	Computes the brain lesions associated to a specific MRI exam.
store_brain_lesions_volume	Stores the computed brain lesions volume in the repository.
compute_volume_value	Computes the volume of brain lesions for a specific MRI exam, DTI metric, and region of interest.
initialize_progress_bar	Initialize the progress bar with the total number of steps.
update_progress_bar	Updates the progress bar by incrementing the current step.

Attributes:

Name	Type	Description
overwrite_database
subjects_repository	SubjectRepository
mri_repository	MRIExamRepository
atlas_repository	AtlasRepository
roi_repository	RegionOfInterestRepository
brain_lesions_volume_repository
dispatcher
current_step
total_steps

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

def __init__(self,
             repositories_registry: RepositoriesRegistry,
             dispatcher: Optional[EventDispatcher] = None,
             overwrite_database: bool = False) -> None:
    """
    Initializes the SegmentDtiAbnormalValues use-case.
    """
    self.overwrite_database = overwrite_database

    self.subjects_repository: SubjectRepository = repositories_registry.get_repository(Subject)
    self.mri_repository: MRIExamRepository = repositories_registry.get_repository(MRIExam)
    self.atlas_repository: AtlasRepository = repositories_registry.get_repository(Atlas)
    self.roi_repository: RegionOfInterestRepository = repositories_registry.get_repository(RegionOfInterest)
    self.brain_lesions_volume_repository = repositories_registry.get_repository(BrainLesionsVolume)

    self.dispatcher = dispatcher
    # Initialize the progress bar attributes
    self.current_step = None
    self.total_steps = None

overwrite_database = overwrite_database instance-attribute

subjects_repository = repositories_registry.get_repository(Subject) instance-attribute

mri_repository = repositories_registry.get_repository(MRIExam) instance-attribute

atlas_repository = repositories_registry.get_repository(Atlas) instance-attribute

roi_repository = repositories_registry.get_repository(RegionOfInterest) instance-attribute

brain_lesions_volume_repository = repositories_registry.get_repository(BrainLesionsVolume) instance-attribute

dispatcher = dispatcher instance-attribute

current_step = None instance-attribute

total_steps = None instance-attribute

compute_all_brain_lesions(dti_metrics, regions_of_interest)

Computes the brain lesions for all patients in the SubjectRepository.

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

def compute_all_brain_lesions(self,
                              dti_metrics: List[DTIMetric],
                              regions_of_interest: List[RegionOfInterest]) -> None:
    """
    Computes the brain lesions for all patients in the SubjectRepository.
    """
    patients = self.subjects_repository.list_all_patients()
    self.initialize_progress_bar(
        number_of_patients=len(patients),
        number_of_dti_metrics=len(dti_metrics),
        number_of_regions_of_interest=len(regions_of_interest) + 1  # +1 for the whole brain
    )
    for patient in patients:
        # Get the MRI exam for the patient
        mri_exam = self.mri_repository.get_exam_for_subject(patient)
        self.compute_brain_lesions_associated_to_mri_exam(mri_exam, dti_metrics, regions_of_interest)

compute_brain_lesions_associated_to_mri_exam(mri_exam, dti_metrics, regions_of_interest)

Computes the brain lesions associated to a specific MRI exam.

Parameters:

Name	Type	Description	Default
mri_exam	MRIExam	The MRI exam for which to compute the brain lesions.	required
dti_metrics	List[DTIMetric]	The DTI metrics for which to compute the brain lesions.	required
regions_of_interest	List[RegionOfInterest]	The supplementary regions of interest to consider for the segmentation. If None, only the whole brain will be considered.	required

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

def compute_brain_lesions_associated_to_mri_exam(self,
                                                 mri_exam: MRIExam,
                                                 dti_metrics: List[DTIMetric],
                                                 regions_of_interest: List[RegionOfInterest]) -> None:
    """
    Computes the brain lesions associated to a specific MRI exam.

    Parameters
    ----------
    mri_exam : MRIExam
        The MRI exam for which to compute the brain lesions.
    dti_metrics : List[DTIMetric]
        The DTI metrics for which to compute the brain lesions.
    regions_of_interest : List[RegionOfInterest]
        The supplementary regions of interest to consider for the segmentation.
        If None, only the whole brain will be considered.
    """
    logger.info(f"Computing brain lesions for MRI exam {mri_exam.id}...")

    # Get the masks for the regions of interest
    all_masks = {}
    for roi in regions_of_interest:
        if roi.name is None:
            raise ValueError("Region of interest must have a name.")
        mask = mri_exam.get_mask(roi)
        all_masks[roi.name] = mask
    all_masks["whole_brain"] = None

    for abnormal_value_type in AbnormalValueType:
        for dti_metric in dti_metrics:
            segmented_dti_map = mri_exam.get_segmented_dti_abnormal_values(dti_metric)
            for region_name, mask in all_masks.items():
                # Create the brain lesions volume with empty value
                brain_lesions_volume = BrainLesionsVolume(
                    mri_exam_id=mri_exam.id,
                    dti_metric=dti_metric,
                    region_of_interest=next((roi for roi in regions_of_interest if roi.name == region_name), None),
                    abnormal_value_type=abnormal_value_type,
                    value_ml=0.0
                )
                if self.brain_lesions_volume_repository.exists(brain_lesions_volume) and not self.overwrite_database:
                    logger.info(f"Brain lesions volume for MRI exam {brain_lesions_volume.mri_exam_id} "
                                   f"and DTI metric {brain_lesions_volume.dti_metric} already exists. "
                                   f"Skipping computation.")
                else:
                    # Compute the volume of brain lesions
                    brain_lesions_volume.value_ml = self.compute_volume_value(
                        segmented_dti_map=segmented_dti_map,
                        region_of_interest_mask=mask,
                        abnormal_value_type=abnormal_value_type
                    )
                    self.store_brain_lesions_volume(brain_lesions_volume)
                    logger.info(f"Successfully computed brain lesions for MRI exam {mri_exam.id}.")
                self.update_progress_bar()

store_brain_lesions_volume(brain_lesions_volume)

Stores the computed brain lesions volume in the repository.

Parameters:

Name	Type	Description	Default
brain_lesions_volume	BrainLesionsVolume	The brain lesions volume to store.	required

Returns:

Type	Description
None

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

def store_brain_lesions_volume(self, brain_lesions_volume: BrainLesionsVolume) -> None:
    """
    Stores the computed brain lesions volume in the repository.

    Parameters
    ----------
    brain_lesions_volume : BrainLesionsVolume
        The brain lesions volume to store.

    Returns
    -------
    None
    """
    if self.overwrite_database:
        self.brain_lesions_volume_repository.update(brain_lesions_volume)
    else:
        if not self.brain_lesions_volume_repository.exists(brain_lesions_volume):
            self.brain_lesions_volume_repository.save(brain_lesions_volume)
        else:
            logger.warning(f"Brain lesions volume for MRI exam {brain_lesions_volume.mri_exam_id} "
                           f"and DTI metric {brain_lesions_volume.dti_metric} already exists. "
                           f"Skipping storage.")

compute_volume_value(segmented_dti_map, region_of_interest_mask, abnormal_value_type) staticmethod

Computes the volume of brain lesions for a specific MRI exam, DTI metric, and region of interest.

Parameters:

Name	Type	Description	Default
segmented_dti_map	DTIAbnormalValues	The segmented DTI abnormal values map for the MRI exam.	required
region_of_interest_mask	Optional[Mask]	The mask defining the region of interest for which to compute the volume. If None, no mask is applied (case of the whole brain).	required
abnormal_value_type	AbnormalValueType	The type of abnormal value to compute the volume for (HIGH of LOW).	required

Returns:

Type	Description
float	The computed volume in milliliters (ml) of the brain lesions.

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

@staticmethod
def compute_volume_value(segmented_dti_map: DTIAbnormalValues,
                         region_of_interest_mask: Optional[Mask],
                         abnormal_value_type: AbnormalValueType) -> float:
    """
    Computes the volume of brain lesions for a specific MRI exam, DTI metric, and region of interest.

    Parameters
    ----------
    segmented_dti_map : DTIAbnormalValues
        The segmented DTI abnormal values map for the MRI exam.
    region_of_interest_mask : Optional[Mask]
        The mask defining the region of interest for which to compute the volume.
        If None, no mask is applied (case of the whole brain).
    abnormal_value_type : AbnormalValueType
        The type of abnormal value to compute the volume for (HIGH of LOW).

    Returns
    -------
    float
        The computed volume in milliliters (ml) of the brain lesions.
    """
    abnormal_voxel_data = cast(AbnormalVoxelData, segmented_dti_map.get_voxel_data())
    abnormal_coordinates = abnormal_voxel_data.get_abnormal_voxels_coordinates(abnormal_value_type)

    if region_of_interest_mask is not None:
        # Filter the coordinates by the region of interest
        true_voxels_maks = region_of_interest_mask.get_true_voxel_coordinates()
        abnormal_coordinates = [coord for coord in abnormal_coordinates if coord in true_voxels_maks]

    # Compute the volume
    return len(abnormal_coordinates) * abnormal_voxel_data.get_voxel_volume_in_ml()

initialize_progress_bar(number_of_patients, number_of_dti_metrics, number_of_regions_of_interest)

Initialize the progress bar with the total number of steps.

Parameters:

Name	Type	Description	Default
number_of_patients	int	The number of patients to process. Each patient will have a number of steps equal to the number of DTI metrics multiplied by the number of regions of interest (plus one for the whole brain).	required

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

def initialize_progress_bar(self,
                            number_of_patients: int,
                            number_of_dti_metrics: int,
                            number_of_regions_of_interest: int,
                            ) -> None:
    """
    Initialize the progress bar with the total number of steps.

    Parameters
    ----------
    number_of_patients : int
        The number of patients to process. Each patient will have a number of steps
        equal to the number of DTI metrics multiplied by the number of regions of interest
        (plus one for the whole brain).
    """
    # Compute the total number of steps
    # *2 because we compute both HIGH and LOW abnormal values
    total_steps = number_of_patients * number_of_dti_metrics * number_of_regions_of_interest * 2

    self.current_step = 0
    self.total_steps = total_steps
    if self.dispatcher is not None:
        self.dispatcher.dispatch(ProgressEvent(0, self.total_steps))

update_progress_bar()

Updates the progress bar by incrementing the current step.

Source code in oxytcmri/domain/use_cases/compute_lesions_volumes.py

def update_progress_bar(self) -> None:
    """
    Updates the progress bar by incrementing the current step.
    """
    if self.dispatcher is not None:
        self.current_step += 1
        self.dispatcher.dispatch(ProgressEvent(self.current_step, self.total_steps))