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 quantile (first quartile)
quartile_75	Calculate the 75th quantile (third quartile)
iqr	Calculate the interquartile range

Attributes:

Name	Type	Description
MEAN_STRATEGY
STD_DEV_STRATEGY
QUARTILE_25_STRATEGY
QUARTILE_75_STRATEGY
IQR_STRATEGY
QUANTILE_5_STRATEGY
QUANTILE_95_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

QUANTILE_5_STRATEGY = StatisticStrategy('quantile 5', quantile_5) class-attribute instance-attribute

QUANTILE_95_STRATEGY = StatisticStrategy('quantile 95', quantile_95) 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

parametric_percentile(values, quantile) staticmethod

Calculate the quantile for a given percentile q.

Parameters:

Name	Type	Description	Default
values	List[float]	The list of values to calculate the quantile on	required
quantile	float	The desired percentile (0-100)	required

Returns:

Type	Description
float	The calculated quantile value

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

@staticmethod
def parametric_percentile(values: List[float], quantile: float) -> float:
    """
    Calculate the quantile for a given percentile q.

    Parameters
    ----------
    values : List[float]
        The list of values to calculate the quantile on
    quantile : float
        The desired percentile (0-100)

    Returns
    -------
    float
        The calculated quantile value
    """
    return float(np.percentile(values, quantile)) 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 StatisticsStrategies.parametric_percentile(values, 25)

quartile_75(values) staticmethod

Calculate 75th quantile 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 quantile with handling for empty lists."""
    return StatisticsStrategies.parametric_percentile(values, 75)

quantile_5(values) staticmethod

Calculate 5th quantile with handling for empty lists.

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

@staticmethod
def quantile_5(values: List[float]) -> float:
    """Calculate 5th quantile with handling for empty lists."""
    return StatisticsStrategies.parametric_percentile(values, 5)

quantile_95(values) staticmethod

Calculate 95th quantile with handling for empty lists.

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

@staticmethod
def quantile_95(values: List[float]) -> float:
    """Calculate 95th quantile with handling for empty lists."""
    return StatisticsStrategies.parametric_percentile(values, 95)

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.QUANTILE_5_STRATEGY,
            cls.QUANTILE_95_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
get_by_parameters	Retrieve a normative value by its parameters.

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

Retrieve a normative value by its parameters.

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

Returns:

Type	Description
NormativeValue	The corresponding normative value

Raises:

Type	Description
EntityIdNotFoundException	If no normative value with the given parameters exists

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

@abstractmethod
def get_by_parameters(self,
                      center: Center,
                      dti_metric: DTIMetric,
                      atlas: Atlas,
                      atlas_label: int,
                      statistic_strategy: StatisticStrategy
                      ) -> NormativeValue:
    """
    Retrieve a normative value by its parameters.

    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

    Returns
    -------
    NormativeValue
        The corresponding normative value

    Raises
    ------
    EntityIdNotFoundException
        If no normative value with the given parameters exists
    """

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: AtlasRepository = repositories_registry.get_repository(Atlas)
    self.centers_repository: CenterRepository = repositories_registry.get_repository(Center)
    self.subjects_repository: SubjectRepository = repositories_registry.get_repository(Subject)
    self.mri_repository: MRIExamRepository = repositories_registry.get_repository(MRIExam)
    self.normative_values_repository: NormativeValueRepository = 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: Optional[List[StatisticStrategy]] = None,
                                 dti_metrics: Optional[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)
        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