`corrct.param_tuning`

Aided regularization parameter estimation.

@author: Nicola VIGANÒ, Computational Imaging group, CWI, The Netherlands, and ESRF - The European Synchrotron, Grenoble, France

Module Contents

Classes

`PerfMeterTask`	Performance tracking class for single reconstructions.
`PerfMeterBatch`	Performance tracking class for batches of reconstructions.
`BaseParameterTuning`	Base class for parameter tuning classes.
`LCurve`	L-curve regularization parameter estimation helper.
`CrossValidation`	Cross-validation hyper-parameter estimation class.

Functions

`format_time`	Convert seconds to a formatted string in the format ::..
`create_random_test_mask`	Create a random mask for cross-validation.
`create_k_fold_test_masks`	Create K random masks for K-fold cross-validation.
`get_lambda_range`	Compute hyper-parameter values within an interval.
`fit_func_min`	Parabolic fit of objective function costs for the different hyper-parameter values.
`_compute_reconstruction`	Compute a single reconstruction.
`_parallel_compute`	Compute reconstructions in parallel.
`_serial_compute`	Compute reconstructions in serial.
`plot_cv_curves`	Plot the relative cross-validation curves for all hyper-parameter values.

Data

`NUM_CPUS`
`MAX_THREADS`
`NDArrayFloat`

API

corrct.param_tuning.NUM_CPUS: None

corrct.param_tuning.MAX_THREADS: ‘int(…)’

corrct.param_tuning.NDArrayFloat: None

corrct.param_tuning.format_time(seconds: float) → str[source]

Convert seconds to a formatted string in the format ::..

Parameters

seconds : float Time in seconds.

Returns

str Formatted time string.

class corrct.param_tuning.PerfMeterTask[source]

Performance tracking class for single reconstructions.

init_time_s: float: None

exec_time_s: float: None

total_time_s: float: None

class corrct.param_tuning.PerfMeterBatch[source]

Performance tracking class for batches of reconstructions.

init_time_s: float: 0.0

proc_time_s: float: 0.0

total_time_s: float: 0.0

tasks_perf: list[corrct.param_tuning.PerfMeterTask]: ‘field(…)’

append(task: corrct.param_tuning.PerfMeterTask) → None[source]

Append a task’s performance metrics to the batch.

Parameters

task : PerfMeterTask The task to append to the batch.

__add__(other: corrct.param_tuning.PerfMeterBatch) → corrct.param_tuning.PerfMeterBatch[source]

Add two PerfMeterBatch instances together.

Parameters

other : PerfMeterBatch The other PerfMeterBatch instance to add to this one.

Returns

PerfMeterBatch A new PerfMeterBatch instance with the sum of the dispatch times, processing times, total times, and concatenated task performance lists from both instances.

__str__() → str[source]

Return a formatted string representation of the performance statistics.

Returns

str Formatted string representation of the performance statistics.

corrct.param_tuning.create_random_test_mask(data_shape: collections.abc.Sequence[int], test_fraction: float = 0.05, dtype: numpy.typing.DTypeLike = np.float32) → corrct.param_tuning.NDArrayFloat[source]

Create a random mask for cross-validation.

Parameters

data_shape : Sequence[int] The shape of the data. test_fraction : float, optional The fraction of pixels to mask. The default is 0.05. dtype : DTypeLike, optional The data type of the mask. The default is np.float32.

Returns

NDArrayFloat The pixel mask.

corrct.param_tuning.create_k_fold_test_masks(data_shape: collections.abc.Sequence[int], k_folds: int, dtype: numpy.typing.DTypeLike = np.float32, seed: int | None = None) → list[numpy.typing.NDArray][source]

Create K random masks for K-fold cross-validation.

Parameters

data_shape : Sequence[int] The shape of the data. k_folds : int The number of folds. dtype : DTypeLike, optional The data type of the masks. The default is np.float32. seed : int | None, optional Seed for the random number generator. The default is None.

Returns

list[NDArray] A list of K pixel masks.

corrct.param_tuning.get_lambda_range(start: float, end: float, num_per_order: int = 4, aligned_order: bool = True) → corrct.param_tuning.NDArrayFloat[source]

Compute hyper-parameter values within an interval.

Parameters

start : float First hyper-parameter value. end : float Last hyper-parameter value. num_per_order : int, optional Number of steps per order of magnitude. The default is 4. aligned_order : bool, optional Whether to align the 1 of each order of magnitude or to the given start value. The default is True.

Returns

NDArrayFloat List of hyper-parameter values.

corrct.param_tuning.fit_func_min(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, f_vals: corrct.param_tuning.NDArrayFloat, f_stds: corrct.param_tuning.NDArrayFloat | None = None, scale: Literal[linear, log] = 'log', verbose: bool = False, plot_result: bool = False) → tuple[float, float][source]

Parabolic fit of objective function costs for the different hyper-parameter values.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat Hyper-parameter values. f_vals : NDArrayFloat Objective function costs of each hyper-parameter value. f_stds : NDArrayFloat, optional Objective function cost standard deviations of each hyper-parameter value. It is only used for plotting purposes. The default is None. scale : str, optional Scale of the fit. Options are: “log” | “linear”. The default is “log”. verbose : bool, optional Whether to produce verbose output, by default False plot_result : bool, optional Whether to plot the result, by default False

Returns

min_hp_val : float Expected hyper-parameter value of the fitted minimum. min_f_val : float Expected objective function cost of the fitted minimum.

corrct.param_tuning._compute_reconstruction(init_fun: collections.abc.Callable | None, exec_fun: collections.abc.Callable[[Any], tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo]], hp_val: float, *, init_fun_kwds: collections.abc.Mapping, exec_fun_kwds: collections.abc.Mapping) → tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo, corrct.param_tuning.PerfMeterTask][source]

Compute a single reconstruction.

Parameters

init_fun : Callable | None The task initialization function. exec_fun : Callable[[Any], tuple[NDArrayFloat, SolutionInfo]] The task execution function. hp_val : float The hyper-parameter value. init_fun_kwds : Mapping Additional keyword arguments to pass to the task initialization function. exec_fun_kwds : Mapping Additional keyword arguments to pass to the task execution function.

Returns

NDArrayFloat The reconstruction. SolutionInfo The solution information. PerfMeterTask The performance statistics for the reconstruction.

corrct.param_tuning._parallel_compute(executor: concurrent.futures.Executor, init_fun: collections.abc.Callable | None, exec_fun: collections.abc.Callable[[Any], tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo]], hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None, verbose: bool = True) → tuple[list[numpy.typing.NDArray], list[corrct.solvers.SolutionInfo], corrct.param_tuning.PerfMeterBatch][source]

Compute reconstructions in parallel.

Parameters

executor : Executor The executor to use for parallel computation. init_fun : Callable | None The task initialization function. exec_fun : Callable[[Any], tuple[NDArrayFloat, SolutionInfo]] The task execution function. hp_vals : float | Sequence[float] | NDArrayFloat A list or array of hyperparameter values to evaluate. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None. verbose : bool, optional Whether to produce verbose output, by default True

Returns

list[NDArray] A list of reconstructions corresponding to each hyperparameter value. list[SolutionInfo] A list of solution information objects corresponding to each reconstruction. PerfMeterBatch A computing performance statistics object containing aggregated performance metrics.

corrct.param_tuning._serial_compute(init_fun: collections.abc.Callable | None, exec_fun: collections.abc.Callable[[Any], tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo]], hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None, verbose: bool = True) → tuple[list[numpy.typing.NDArray], list[corrct.solvers.SolutionInfo], corrct.param_tuning.PerfMeterBatch][source]

Compute reconstructions in serial.

Parameters

init_fun : Callable | None The task initialization function. exec_fun : Callable[[Any], tuple[NDArrayFloat, SolutionInfo]] The task execution function. hp_vals : float | Sequence[float] | NDArrayFloat A list or array of hyperparameter values to evaluate. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None. verbose : bool, optional Whether to produce verbose output, by default True

Returns

list[NDArray] A list of reconstructions corresponding to each hyperparameter value. list[SolutionInfo] A list of solution information objects corresponding to each reconstruction. PerfMeterBatch A computing performance statistics object containing aggregated performance metrics.

corrct.param_tuning.plot_cv_curves(solution_infos: list[corrct.solvers.SolutionInfo], hp_vals: collections.abc.Sequence[float]) → None[source]

Plot the relative cross-validation curves for all hyper-parameter values.

Parameters

solution_infos : list[SolutionInfo] List of SolutionInfo objects containing the relative cross-validation residuals. hp_vals : Sequence[float] Sequence of hyper-parameter values corresponding to the solution_infos.

class corrct.param_tuning.BaseParameterTuning(dtype: numpy.typing.DTypeLike = np.float32, parallel_eval: concurrent.futures.Executor | int | bool = True, verbose: bool = False, plot_result: bool = False, print_timings: bool = False)[source]

Bases: abc.ABC

Base class for parameter tuning classes.

Initialization

Initialize a base helper class.

Parameters

dtype : DTypeLike, optional Type of the data, by default np.float32 parallel_eval : Executor | int | bool, optional Whether to evaluate results in parallel, by default True verbose : bool, optional Whether to produce verbose output, by default False plot_result : bool, optional Whether to plot the results, by default False print_timings : bool, optional Whether to print the performance metrics, by default False

_task_init_function: collections.abc.Callable | None: None

_task_exec_function: collections.abc.Callable[[Any], tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo]] | None: None

parallel_eval: int | concurrent.futures.Executor: None

dtype: numpy.typing.DTypeLike: None

verbose: bool: None

plot_result: bool: None

print_timings: bool: None

property task_init_function: collections.abc.Callable | None

Return the local reference to the task initialization function.

Returns

Callable | None The task initialization function, if initialized.

property task_exec_function: collections.abc.Callable[[Any], tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo]]

Return the local reference to the task execution function.

Returns

Callable[[Any], tuple[NDArrayFloat, SolutionInfo]] The task execution function.

property solver_spawning_function: collections.abc.Callable | None: Return the local reference to the task initialization function.

property solver_calling_function: collections.abc.Callable[[Any], tuple[corrct.param_tuning.NDArrayFloat, corrct.solvers.SolutionInfo]]: Return the local reference to the task execution function.

static get_lambda_range(start: float, end: float, num_per_order: int = 4) → corrct.param_tuning.NDArrayFloat[source]

Compute regularization weights within an interval.

Parameters

start : float First regularization weight. end : float Last regularization weight. num_per_order : int, optional Number of steps per order of magnitude. The default is 4.

Returns

NDArrayFloat List of regularization weights.

process_hp_vals(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None) → tuple[list[numpy.typing.NDArray], list[corrct.solvers.SolutionInfo], corrct.param_tuning.PerfMeterBatch][source]

Compute reconstructions, solution information, and computing performance statistics for all hyperparameter values.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat A list or array of hyperparameter values to evaluate. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None.

Returns

tuple[list[NDArray], list[SolutionInfo], PerfStatsBatch] A tuple containing: - A list of reconstructions corresponding to each hyperparameter value. - A list of solution information objects corresponding to each reconstruction. - A computing performance statistics object containing aggregated performance metrics.

compute_reconstruction_error(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, gnd_truth: corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None) → tuple[corrct.param_tuning.NDArrayFloat, corrct.param_tuning.NDArrayFloat][source]

Compute the reconstruction errors for each hyper-parameter values against the ground truth.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat List of hyper-parameter values. gnd_truth : NDArrayFloat Expected reconstruction. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None.

Returns

err_l1 : NDArrayFloat l1-norm errors for each reconstruction. err_l2 : NDArrayFloat l2-norm errors for each reconstruction.

abstract compute_loss_values(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None, return_all: bool = False) → corrct.param_tuning.NDArrayFloat | tuple[corrct.param_tuning.NDArrayFloat, list[corrct.param_tuning.NDArrayFloat], list[corrct.solvers.SolutionInfo], corrct.param_tuning.PerfMeterBatch][source]

Compute the objective function costs for a list of hyper-parameter values.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat List of hyper-parameter values. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None. return_all : bool, optional If True, return all the computation information (reconstructions, all loss values, performance). Default is False.

Returns

NDArrayFloat Objective function cost for each hyper-parameter value. recs : NDArrayFloat, optional Reconstructions for each hyper-parameter value (returned only if return_all is True). recs_info : list[SolutionInfo], optional Solution information for each reconstruction (returned only if return_all is True). perf_batch : PerfStatsBatch, optional Performance statistics for the batch of reconstructions (returned only if return_all is True).

class corrct.param_tuning.LCurve(loss_function: collections.abc.Callable, data_dtype: numpy.typing.DTypeLike = np.float32, parallel_eval: concurrent.futures.Executor | int | bool = True, verbose: bool = False, plot_result: bool = False, print_timings: bool = False)[source]

Bases: corrct.param_tuning.BaseParameterTuning

L-curve regularization parameter estimation helper.

Initialization

Create an L-curve regularization parameter estimation helper.

Parameters

loss_function : Callable The loss function for the computation of the L-curve values. data_dtype : DTypeLike, optional Type of the input data. The default is np.float32. parallel_eval : Executor | int | bool, optional Compute loss and error values in parallel. The default is True. verbose : bool, optional Print verbose output. The default is False. plot_result : bool, optional Plot results. The default is False. print_timings : bool, optional Whether to print the performance metrics, by default False

compute_loss_values(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None, return_all: bool = False) → corrct.param_tuning.NDArrayFloat | tuple[corrct.param_tuning.NDArrayFloat, list[corrct.param_tuning.NDArrayFloat], list[corrct.solvers.SolutionInfo], corrct.param_tuning.PerfMeterBatch][source]

Compute the objective function costs for a list of hyper-parameter values.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat List of hyper-parameter values. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None. return_all : bool, optional If True, return all the computation information (reconstructions, all loss values, performance). Default is False.

Returns

f_vals : NDArrayFloat Objective function cost for each hyper-parameter value. recs : list[NDArrayFloat], optional Reconstructions for each hyper-parameter value (returned only if return_all is True). recs_info : list[SolutionInfo], optional Solution information for each reconstruction (returned only if return_all is True). perf_batch : PerfStatsBatch, optional Performance statistics for the batch of reconstructions (returned only if return_all is True).

class corrct.param_tuning.CrossValidation(data_shape: collections.abc.Sequence[int], cv_fraction: float | None = 0.1, num_averages: int = 5, mask_param_name: str = 'b_test_mask', parallel_eval: concurrent.futures.Executor | int | bool = True, dtype: numpy.typing.DTypeLike = np.float32, verbose: bool = False, plot_result: bool = False, print_timings: bool = False)[source]

Bases: corrct.param_tuning.BaseParameterTuning

Cross-validation hyper-parameter estimation class.

Initialization

Create a cross-validation hyper-parameter estimation helper.

Parameters

data_shape : Sequence[int] Shape of the projection data. cv_fraction : float | None, optional Fraction of detector points to use for the leave-out set. If None, K-fold cross-validation is used, where num_averages indicates the number of k-folds. The default is 0.1. num_averages : int, optional Number of averages random leave-out sets to use. The default is 5. mask_param_name: str, optional The parameter name in the task execution function that accepts the data masks. The default is “b_test_mask”. parallel_eval : Executor | int | bool, optional Compute loss and error values in parallel. The default is True. dtype : DTypeLike, optional Type of the input data. The default is np.float32. verbose : bool, optional Print verbose output. The default is False. plot_result : bool, optional Plot results. The default is False. print_timings : bool, optional Whether to print the performance metrics, by default False

data_shape: collections.abc.Sequence[int]: None

cv_fraction: float | None: None

num_averages: int: None

data_cv_masks: list[numpy.typing.NDArray]: None

mask_param_name: str: None

compute_loss_values(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, *, init_fun_kwds: collections.abc.Mapping | None = None, exec_fun_kwds: collections.abc.Mapping | None = None, return_all: bool = False) → tuple[corrct.param_tuning.NDArrayFloat, corrct.param_tuning.NDArrayFloat, list[corrct.param_tuning.NDArrayFloat]] | tuple[corrct.param_tuning.NDArrayFloat, corrct.param_tuning.NDArrayFloat, list[tuple[list[corrct.param_tuning.NDArrayFloat], list[corrct.solvers.SolutionInfo], corrct.param_tuning.PerfMeterBatch]]][source]

Compute objective function values for all requested hyper-parameter values.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat Hyper-parameter values (e.g., regularization weight) to evaluate. init_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task initialization function. By default None. exec_fun_kwds : Mapping | None, optional Additional keyword arguments to pass to the task execution function. By default None. return_all : bool, optional If True, return the reconstructions along with the loss values. Default is False.

Returns

f_avgs : NDArrayFloat Average objective function costs for each hyper-parameter value. f_stds : NDArrayFloat Standard deviation of objective function costs for each hyper-parameter value. f_vals : NDArrayFloat Objective function costs for each hyper-parameter value. recs : list[NDArrayFloat], optional Reconstructions for each hyper-parameter value (returned only if return_all is True).

fit_loss_min(hp_vals: float | collections.abc.Sequence[float] | corrct.param_tuning.NDArrayFloat, f_vals: corrct.param_tuning.NDArrayFloat, f_stds: corrct.param_tuning.NDArrayFloat | None = None, scale: Literal[linear, log] = 'log') → tuple[float, float][source]

Parabolic fit of objective function costs for the different hyper-parameter values.

Parameters

hp_vals : float | Sequence[float] | NDArrayFloat Hyper-parameter values. f_vals : NDArrayFloat Objective function costs of each hyper-parameter value. f_stds : NDArrayFloat, optional Objective function cost standard deviations of each hyper-parameter value. It is only used for plotting purposes. The default is None. scale : str, optional Scale of the fit. Options are: “log” | “linear”. The default is “log”.

Returns

min_hp_val : float Expected hyper-parameter value of the fitted minimum. min_f_val : float Expected objective function cost of the fitted minimum.

corrct.param_tuning

Module Contents

Classes

Functions

Data

API

`corrct.param_tuning`