compute_dti_normative_values

oxytcmri.domain.use_cases.compute_dti_normative_values

Classes:

Name	Description
StatisticStrategy	A statistical calculation strategy that encapsulates both the type
StatisticsStrategies	A collection of statistical calculation strategies for DTI metrics.
NormativeValue	Represents a normative value for a DTI metric.
NormativeValueRepository	Abstract base class for Normative Value repository.
ComputeDTINormativeValues	Use case for computing normative DTI values from healthy volunteers.

StatisticStrategy(name, calculate) dataclass

A statistical calculation strategy that encapsulates both the type and the calculation method.

Parameters:

Name	Type	Description	Default
name	str	A human-readable name for the statistical strategy	required
calculate	Callable[[List[float]], float]	A function that calculates a specific statistical measure	required

Methods:

Name	Description
__call__	Allows the strategy to be called directly like a function

Attributes:

Name	Type	Description
name	str
calculate	Callable[[List[float]], float]

name instance-attribute

calculate instance-attribute

StatisticsStrategies

A collection of statistical calculation strategies for DTI metrics.

This class defines various statistical methods that can be applied to lists of float values, with built-in handling for empty lists.

Methods:

Name	Description
mean	Calculate the arithmetic mean of the values
std_dev	Calculate the standard deviation of the values
quartile_25	Calculate the 25th percentile (first quartile)
quartile_75	Calculate the 75th percentile (third quartile)
iqr	Calculate the interquartile range

Attributes:

Name	Type	Description
MEAN_STRATEGY
STD_DEV_STRATEGY
QUARTILE_25_STRATEGY
QUARTILE_75_STRATEGY
IQR_STRATEGY

MEAN_STRATEGY = StatisticStrategy('mean', mean) class-attribute instance-attribute

STD_DEV_STRATEGY = StatisticStrategy('standard deviation', std_dev) class-attribute instance-attribute

QUARTILE_25_STRATEGY = StatisticStrategy('quartile 25', quartile_25) class-attribute instance-attribute

QUARTILE_75_STRATEGY = StatisticStrategy('quartile 75', quartile_75) class-attribute instance-attribute

IQR_STRATEGY = StatisticStrategy('interquartile range', iqr) class-attribute instance-attribute

mean(values) staticmethod

Calculate mean with handling for empty lists.

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

@staticmethod
def mean(values: List[float]) -> float:
    """Calculate mean with handling for empty lists."""
    return float(np.mean(values)) if values else 0.0

std_dev(values) staticmethod

Calculate standard deviation with handling for empty lists.

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

@staticmethod
def std_dev(values: List[float]) -> float:
    """Calculate standard deviation with handling for empty lists."""
    return float(np.std(values)) if values else 0.0

quartile_25(values) staticmethod

Calculate 25th percentile with handling for empty lists.

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

@staticmethod
def quartile_25(values: List[float]) -> float:
    """Calculate 25th percentile with handling for empty lists."""
    return float(np.percentile(values, 25)) if values else 0.0

quartile_75(values) staticmethod

Calculate 75th percentile with handling for empty lists.

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

@staticmethod
def quartile_75(values: List[float]) -> float:
    """Calculate 75th percentile with handling for empty lists."""
    return float(np.percentile(values, 75)) if values else 0.0

iqr(values) staticmethod

Calculate interquartile range with handling for empty lists.

Returns the difference between 75th and 25th percentiles.

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

@staticmethod
def iqr(values: List[float]) -> float:
    """
    Calculate interquartile range with handling for empty lists.

    Returns the difference between 75th and 25th percentiles.
    """
    if not values:
        return 0.0
    q1 = np.percentile(values, 25)
    q3 = np.percentile(values, 75)
    return float(q3 - q1)

all() classmethod

Retrieve all defined statistical strategies.

Returns:

Type	Description
List[StatisticStrategy]	A list of all available statistical strategies

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

@classmethod
def all(cls):
    """
    Retrieve all defined statistical strategies.

    Returns
    -------
    List[StatisticStrategy]
        A list of all available statistical strategies
    """
    return [cls.MEAN_STRATEGY, cls.STD_DEV_STRATEGY, cls.QUARTILE_25_STRATEGY, cls.QUARTILE_75_STRATEGY,
            cls.IQR_STRATEGY]

get_by_name(name) classmethod

Retrieve a statistical strategy by its name.

Parameters:

Name	Type	Description	Default
name	str	The name of the statistical strategy	required

Returns:

Type	Description
StatisticStrategy	The corresponding statistical strategy

Raises:

Type	Description
ValueError	If no strategy with the given name exists

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

@classmethod
def get_by_name(cls, name: str) -> StatisticStrategy:
    """
    Retrieve a statistical strategy by its name.

    Parameters
    ----------
    name : str
        The name of the statistical strategy

    Returns
    -------
    StatisticStrategy
        The corresponding statistical strategy

    Raises
    ------
    ValueError
        If no strategy with the given name exists
    """
    for strategy in cls.all():
        if strategy.name == name:
            return strategy
    raise ValueError(f"No statistical strategy found with name: {name}")

NormativeValue(center, dti_metric, atlas, atlas_label, statistic_strategy, value) dataclass

Represents a normative value for a DTI metric.

A normative value is a statistical measure calculated from healthy subjects for a specific DTI metric in a specific anatomical region.

Parameters:

Name	Type	Description	Default
center	Center	The medical center where the normative value was calculated	required
dti_metric	DTIMetric	The type of DTI metric (e.g., MD, FA)	required
atlas	Atlas	The atlas used for segmentation	required
atlas_label	int	The specific label within the atlas	required
statistic_strategy	StatisticStrategy	The statistical strategy used to compute the value	required
value	float	The calculated statistical value	required

Attributes:

Name	Type	Description
center	Center
dti_metric	DTIMetric
atlas	Atlas
atlas_label	int
statistic_strategy	StatisticStrategy
value	float

center instance-attribute

dti_metric instance-attribute

atlas instance-attribute

atlas_label instance-attribute

statistic_strategy instance-attribute

value instance-attribute

NormativeValueRepository

Bases: Repository[NormativeValue, None], ABC

Abstract base class for Normative Value repository.

Defines the interface for saving and retrieving normative values.

Methods:

Name	Description
exists	Check if a normative value has already been computed with the given attributes.

exists(center, dti_metric, atlas, atlas_label, statistic_strategy) abstractmethod

Check if a normative value has already been computed with the given attributes.

Parameters:

Name	Type	Description	Default
center	Center	The medical center where the normative value was calculated	required
dti_metric	DTIMetric	The type of DTI metric (e.g., MD, FA)	required
atlas	Atlas	The atlas used for segmentation	required
atlas_label	int	The specific label within the atlas	required
statistic_strategy	StatisticStrategy	The statistical strategy used to compute the value	required

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

@abstractmethod
def exists(self,
           center: Center,
           dti_metric: DTIMetric,
           atlas: Atlas,
           atlas_label: int,
           statistic_strategy: StatisticStrategy
           ) -> bool:
    """
    Check if a normative value has already been computed with the given attributes.

    Parameters
    ----------
    center : Center
        The medical center where the normative value was calculated
    dti_metric : DTIMetric
        The type of DTI metric (e.g., MD, FA)
    atlas : Atlas
        The atlas used for segmentation
    atlas_label : int
        The specific label within the atlas
    statistic_strategy : StatisticStrategy
        The statistical strategy used to compute the value
    """

ComputeDTINormativeValues(repositories_registry, dispatcher=None)

Use case for computing normative DTI values from healthy volunteers.

This class orchestrates the process of extracting DTI values from healthy subjects and computing various statistical measures.

Attributes:

Name	Type	Description
subjects_repository	SubjectRepository	Repository for accessing subject information
mri_repository	MRIExamRepository	Repository for accessing MRI data
atlas_repository	AtlasRepository	Repository for accessing atlas information
centers_repository	CenterRepository	Repository for accessing center information

Initialize the use case with a subject repository.

Parameters:

Name	Type	Description	Default
repositories_registry	RepositoriesRegistry	Registry for accessing various repositories	required
dispatcher	EventDispatcher	Event dispatcher for dispatching events (progress bar, logs, etc.)	None

Methods:

Name	Description
compute_total_steps	Get the total number of steps for the progress bar.
initialize_progress_bar	Initialize the progress bar using the dispatcher.
update_progress_bar	Update the progress bar using the dispatcher.
compute_all_normative_values	Compute normative values for all DTI metrics, statistical strategies, centers, atlases, and labels.
process_center	Process a specific center for a given DTI metric and statistical strategy.
collect_dti_values_for_region	Collect DTI values for a specific region from multiple subjects.
get_regions_of_interest	Retrieve the list of regions of interest.
store_normative_value	Store the normative value in the repository.

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

def __init__(
        self,
        repositories_registry: RepositoriesRegistry,
        dispatcher: EventDispatcher = None
) -> None:
    """
    Initialize the use case with a subject repository.

    Parameters
    ----------
    repositories_registry: RepositoriesRegistry
        Registry for accessing various repositories
    dispatcher : EventDispatcher, optional
        Event dispatcher for dispatching events (progress bar, logs, etc.)
    """
    # repositories
    self.atlas_repository = repositories_registry.get_repository(Atlas)
    self.centers_repository = repositories_registry.get_repository(Center)
    self.subjects_repository = repositories_registry.get_repository(Subject)
    self.mri_repository = repositories_registry.get_repository(MRIExam)
    self.normative_values_repository = repositories_registry.get_repository(NormativeValue)

    # progress bar
    self.dispatcher = dispatcher
    self.current_step = None
    self.total_steps = None

atlas_repository = repositories_registry.get_repository(Atlas) instance-attribute

centers_repository = repositories_registry.get_repository(Center) instance-attribute

subjects_repository = repositories_registry.get_repository(Subject) instance-attribute

mri_repository = repositories_registry.get_repository(MRIExam) instance-attribute

normative_values_repository = repositories_registry.get_repository(NormativeValue) instance-attribute

dispatcher = dispatcher instance-attribute

current_step = None instance-attribute

total_steps = None instance-attribute

compute_total_steps(dti_metrics_count, statistics_strategies_count)

Get the total number of steps for the progress bar.

Returns:

Type	Description
int	The total number of steps

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

def compute_total_steps(self,
                        dti_metrics_count: int,
                        statistics_strategies_count: int
                        ) -> int:
    """
    Get the total number of steps for the progress bar.

    Returns
    -------
    int
        The total number of steps
    """
    centers = self.centers_repository.list_all()

    # Get count of all atlas labels
    atlas_labels_count = 0
    atlases = self.atlas_repository.list_all()
    for atlas in atlases:
        atlas_labels_count += len(atlas.labels)

    return len(centers) * atlas_labels_count * dti_metrics_count * statistics_strategies_count

initialize_progress_bar(dti_metrics_to_process, statistics_strategies_to_process)

Initialize the progress bar using the dispatcher.

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

def initialize_progress_bar(self,
                            dti_metrics_to_process: List[DTIMetric],
                            statistics_strategies_to_process: List[StatisticStrategy]
                            ) -> None:
    """
    Initialize the progress bar using the dispatcher.
    """
    self.current_step = 0
    self.total_steps = self.compute_total_steps(len(dti_metrics_to_process),
                                           len(statistics_strategies_to_process))
    if self.dispatcher is not None:
        self.dispatcher.dispatch(ProgressEvent(0, self.total_steps))

update_progress_bar()

Update the progress bar using the dispatcher.

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

def update_progress_bar(self) -> None:
    """
    Update the progress bar using the dispatcher.
    """
    if self.dispatcher is not None:
        self.current_step += 1
        self.dispatcher.dispatch(ProgressEvent(self.current_step, self.total_steps))

compute_all_normative_values(statistics_strategies=None, dti_metrics=None)

Compute normative values for all DTI metrics, statistical strategies, centers, atlases, and labels.

This is the main entry point that orchestrates the computation process.

Parameters:

Name	Type	Description	Default
statistics_strategies	List[StatisticStrategy]	List of statistical strategies to include in computations. If provided, only this subset will be computed.	None
dti_metrics	List[DTIMetric]	List of DTI metrics to include in computations. If provided, only this subset will be computed.	None

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

def compute_all_normative_values(self,
                                 statistics_strategies: List[StatisticStrategy] = None,
                                 dti_metrics: List[DTIMetric] = None
                                 ) -> None:
    """
    Compute normative values for all DTI metrics, statistical strategies, centers, atlases, and labels.

    This is the main entry point that orchestrates the computation process.

    Parameters
    ----------
    statistics_strategies : List[StatisticStrategy], optional
        List of statistical strategies to include in computations.
        If provided, only this subset will be computed.
    dti_metrics : List[DTIMetric], optional
        List of DTI metrics to include in computations.
        If provided, only this subset will be computed.
    """
    # Get all resources
    dti_metrics_to_process = dti_metrics or list(DTIMetric)
    statistic_strategies_to_process = statistics_strategies or StatisticsStrategies.all()
    centers = self.centers_repository.list_all()
    regions_of_interest = self.get_regions_of_interest()

    # Initialize progress bar
    self.initialize_progress_bar(
        dti_metrics_to_process,
        statistic_strategies_to_process,
    )

    # Process each DTI metric
    for dti_metric in dti_metrics_to_process:
        for statistic_strategy in statistic_strategies_to_process:
            for region_of_interest in regions_of_interest:
                for center in centers:
                    self.process_center(center, dti_metric, statistic_strategy, region_of_interest)

process_center(center, dti_metric, statistic_strategy, region_of_interest)

Process a specific center for a given DTI metric and statistical strategy.

Parameters:

Name	Type	Description	Default
center	Center	The center to process	required
dti_metric	DTIMetric	The DTI metric to process	required
statistic_strategy	StatisticStrategy	The statistical strategy to apply	required
region_of_interest	RegionOfInterest	The region of interest to process	required

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

def process_center(self,
                   center: Center,
                   dti_metric: DTIMetric,
                   statistic_strategy: StatisticStrategy,
                   region_of_interest: RegionOfInterest) -> None:
    """
    Process a specific center for a given DTI metric and statistical strategy.

    Parameters
    ----------
    center : Center
        The center to process
    dti_metric : DTIMetric
        The DTI metric to process
    statistic_strategy : StatisticStrategy
        The statistical strategy to apply
    region_of_interest : RegionOfInterest
        The region of interest to process
    """
    logger = getLogger(__name__)
    logger.info(f"Processing center {center.name} for {dti_metric.name} with {statistic_strategy.name} "
                f"in {region_of_interest.atlas.name} region of interest with label {region_of_interest.labels}")

    if len(region_of_interest.labels) > 1:
        raise ValueError("Region of interest must have only one label in this context")

    label_of_interest = region_of_interest.labels[0]
    # Check if the normative value already exists
    if self.normative_values_repository.exists(
            center, dti_metric, region_of_interest.atlas, label_of_interest, statistic_strategy):
        logger.info(f"Normative value {center.name}-{dti_metric.name}-{region_of_interest.atlas.name}"
                    f"-{label_of_interest}-{statistic_strategy.name}already exists, skipping")
        # Skip if the normative value already exists
        self.update_progress_bar()
        return

    # Get healthy subjects for this center
    healthy_subjects = self.subjects_repository.find_subjects_by_center(
        center=center, subject_type=SubjectType.HEALTHY_VOLUNTEER
    )

    if not healthy_subjects:
        # No healthy subjects for this center, skip
        return

    # Collect DTI values for this region
    all_dti_values = self.collect_dti_values_for_region(
        healthy_subjects, dti_metric, region_of_interest)

    if not all_dti_values:
        logger.info(f"No DTI values found for {center.name} {region_of_interest.atlas.name} "
                    f"{label_of_interest}, skipping")
        self.update_progress_bar()
        return

    # Calculate the statistical value
    statistics_value = statistic_strategy(all_dti_values)

    # Create and save the normative value
    normative_value = NormativeValue(
        center=center,
        dti_metric=dti_metric,
        atlas=region_of_interest.atlas,
        atlas_label=label_of_interest,
        statistic_strategy=statistic_strategy,
        value=statistics_value
    )

    self.store_normative_value(normative_value)
    self.update_progress_bar()

collect_dti_values_for_region(subjects, dti_metric, region_of_interest)

Collect DTI values for a specific region from multiple subjects.

Parameters:

Name	Type	Description	Default
subjects	List[Subject]	List of subjects to extract values from	required
dti_metric	DTIMetric	The DTI metric to extract	required
region_of_interest	RegionOfInterest	The region of interest to extract values from	required

Returns:

Type	Description
List[float]	Combined list of DTI values from all subjects

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

def collect_dti_values_for_region(
        self,
        subjects: List[Subject],
        dti_metric: DTIMetric,
        region_of_interest: RegionOfInterest
) -> List[float]:
    """
    Collect DTI values for a specific region from multiple subjects.

    Parameters
    ----------
    subjects : List[Subject]
        List of subjects to extract values from
    dti_metric : DTIMetric
        The DTI metric to extract
    region_of_interest : RegionOfInterest
        The region of interest to extract values from

    Returns
    -------
    List[float]
        Combined list of DTI values from all subjects
    """
    all_dti_values = []

    for subject in subjects:
        # Extract DTI values for the subject
        mri_exam = self.mri_repository.get_exam_for_subject(subject.id)
        values = mri_exam.extract_dti_values_for_region(dti_metric, region_of_interest)

        if values:
            all_dti_values.extend(values)

    return all_dti_values

get_regions_of_interest()

Retrieve the list of regions of interest.

Returns:

Type	Description
List[RegionOfInterest]	A list of regions of interest

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

def get_regions_of_interest(self) -> List[RegionOfInterest]:
    """
    Retrieve the list of regions of interest.

    Returns
    -------
    List[RegionOfInterest]
        A list of regions of interest
    """
    result = []
    for atlas in self.atlas_repository.list_all():
        for atlas_label in atlas.labels:
            roi = RegionOfInterest(atlas=atlas, labels=[atlas_label])
            result.append(roi)
    return result

store_normative_value(normative_value)

Store the normative value in the repository.

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

def store_normative_value(self, normative_value: NormativeValue) -> None:
    """Store the normative value in the repository."""
    try:
        self.normative_values_repository.save(normative_value)
    except Exception as error:
        logger = getLogger(__name__)
        logger.error(f"Error storing normative value {normative_value}: {error}", exc_info=True)
        # Skip storing if an error occurs
        return