Configuration¶

Configuration management, the immutable parameter registry, parameter space definitions with dual prior system, and shared type aliases.

Config Manager¶

Configuration manager for heterodyne analysis.

exception heterodyne.config.manager.ConfigurationError[source]

Bases: Exception

Raised when configuration is invalid.

class heterodyne.config.manager.ConfigManager[source]

Bases: object

Manager for heterodyne analysis configuration.

Handles loading, validation, and access to configuration settings.

__init__(config)[source]

Initialize with configuration dictionary.

Parameters:: config (dict[str, Any]) – Configuration dictionary

classmethod from_yaml(path)[source]

Load configuration from YAML file.

Parameters:: path (Path | str) – Path to YAML file
Return type:: ConfigManager
Returns:: ConfigManager instance

classmethod from_dict(config)[source]

Create from dictionary.

Parameters:: config (dict[str, Any]) – Configuration dictionary
Return type:: ConfigManager
Returns:: ConfigManager instance

classmethod from_json(path)[source]

Load configuration from JSON file.

Parameters:: path (Path | str) – Path to JSON file
Return type:: ConfigManager
Returns:: ConfigManager instance

property raw_config: dict[str, Any]: Get raw configuration dictionary (deep copy to prevent mutation).

property data_file_path: Path: Path to experimental data file.

property data_folder_path: Path | None: Optional folder path for data.

property file_format: str: Data file format.

property cache_file_path: Path | None: Directory for cache files (falls back to data_folder_path, then None).

property cache_filename_template: str | None: Template for cache filenames with ${variable} substitution.

property cache_compression: bool: Whether to compress cache files.

property dt: float: Time step [seconds].

property start_frame: int: Starting frame (1-indexed).

property end_frame: int: Ending frame (1-indexed, inclusive).

property time_length: int: Number of time points (derived from frame range).

property t_start: int: Starting time index, 0-indexed (derived from start_frame).

property wavevector_q: float: Scattering wavevector magnitude [Å⁻¹].

property phi_angles: list[float] | None: List of phi angles for analysis.

property stator_rotor_gap: float | None: Stator-rotor gap [Å] (optional geometry metadata).

property parameters_config: dict[str, Any]: Get parameters configuration section.

get_parameter_value(group, name)[source]

Get a specific parameter value.

Parameters:

group (str) – Parameter group (‘reference’, ‘sample’, etc.)
name (str) – Parameter name within group

Return type:

float

Returns:

Parameter value

get_parameter_vary(group, name)[source]

Check if parameter varies in optimization.

Parameters:

group (str) – Parameter group
name (str) – Parameter name

Return type:

bool

Returns:

Whether parameter varies

property optimization_method: str: Optimization method (‘nlsq’ or ‘cmc’).

property nlsq_config: dict[str, Any]: NLSQ optimization settings (returns a copy to prevent mutation).

property cmc_config: dict[str, Any]: CMC analysis settings (returns a copy to prevent mutation).

update_optimization_config(section, key, value)[source]

Update a single optimization config key in-place.

Parameters:

section (str) – Optimization sub-section (“nlsq” or “cmc”).
key (str) – Configuration key to update.
value (Any) – New value for the key.

Return type:

None

get_config()[source]

Return a deep copy of the raw configuration dictionary.

Return type:: dict[str, Any]
Returns:: Deep copy of the full configuration dictionary

get_cmc_config()[source]

Return merged CMC config with defaults applied.

Return type:: dict[str, Any]
Returns:: CMC configuration with defaults merged in

property output_dir: Path: Output directory path.

to_yaml(path)[source]

Save configuration to YAML file.

Parameters:: path (Path | str) – Output path
Return type:: None

heterodyne.config.manager.load_xpcs_config(path)[source]

Load XPCS analysis configuration from file.

Convenience function for loading configuration.

Parameters:: path (Path | str) – Path to YAML configuration file
Return type:: ConfigManager
Returns:: ConfigManager instance

Parameter Registry¶

Immutable MappingProxyType-based registry defining all 16 parameters (14 physics + 2 scaling) with bounds, defaults, and prior specifications.

Parameter registry with metadata and bounds for heterodyne model.

All length units use Å (angstroms) for consistency with practical XPCS convention.

class heterodyne.config.parameter_registry.ParameterInfo[source]

Bases: object

Metadata for a single model parameter.

name: Parameter name matching canonical order in parameter_names.py.

default: Default value (must be within [min_bound, max_bound]).

min_bound: Lower bound for optimization.

max_bound: Upper bound for optimization.

description: Human-readable description.

unit: Physical unit string (e.g. “Å²/s^α”).

group: Parameter group name.

vary_default: Whether this parameter varies by default in optimization.

log_space: If True, MCMC samplers should reparameterize in log-space.

prior_mean: Center of the default Bayesian prior (None = midpoint of bounds).

prior_std: Width of the default Bayesian prior (None = half-range of bounds).

is_scaling: If True, this parameter participates in per-angle expansion.

is_physical: If True, this is a physical model parameter (not scaling).

is_flow: If True, this parameter is related to flow/velocity.

name: str

default: float

min_bound: float

max_bound: float

description: str

unit: str = ''

group: str = ''

vary_default: bool = True

log_space: bool = False

prior_mean: float | None = None

prior_std: float | None = None

is_scaling: bool = False

is_physical: bool = True

is_flow: bool = False

validate_value(value)[source]

Check if value is within bounds.

Return type:: bool

clip_value(value)[source]

Clip value to bounds.

Return type:: float

__init__(name, default, min_bound, max_bound, description, unit='', group='', vary_default=True, log_space=False, prior_mean=None, prior_std=None, is_scaling=False, is_physical=True, is_flow=False)

class heterodyne.config.parameter_registry.ParameterRegistry[source]

Bases: object

Registry of all heterodyne model parameters with metadata.

__post_init__()[source]

Ensure parameters are immutable.

Return type:: None

__getitem__(name)[source]

Get parameter info by name.

Return type:: ParameterInfo

__iter__()[source]

Iterate over parameter names in canonical order (14 physics + 2 scaling).

Return type:: Iterator[str]

get_defaults()[source]

Get default values for all parameters.

Return type:: dict[str, float]

get_bounds()[source]

Get (lower_bounds, upper_bounds) as lists.

Return type:: tuple[list[float], list[float]]

get_group(group_name)[source]

Get all parameters in a group.

Return type:: list[ParameterInfo]

get_varying_indices(vary_flags)[source]

Get indices of parameters that vary in optimization.

Return type:: list[int]

get_log_space_names()[source]

Get names of parameters that should be sampled in log-space.

Return type:: list[str]

get_scaling_names()[source]

Get names of per-angle scaling parameters.

Return type:: list[str]

__init__(_parameters=<factory>)

Parameter Manager¶

Parameter manager for heterodyne model optimization.

class heterodyne.config.parameter_manager.BoundDict[source]

Bases: TypedDict

Bound specification for a single parameter.

name: str

min: float

max: float

type: str

class heterodyne.config.parameter_manager.ParameterManager[source]

Bases: object

Manages parameter values, constraints, and transformations.

Provides the bridge between configuration and optimization by: - Managing which parameters vary vs are fixed - Handling parameter transformations (e.g., bounded -> unbounded) - Constructing full parameter arrays from varying subsets - Validating parameter values against physics constraints

Performance caching is enabled by default for repeated bound and active-parameter queries.

space: ParameterSpace

__post_init__()[source]

Build default bounds lookup from the registry, then merge config overrides.

Return type:: None

property n_params: int: Total number of physics model parameters (14).

property n_varying: int: Number of physics parameters that vary in optimization.

property varying_names: list[str]: Names of varying physics parameters (excludes scaling).

property varying_indices: list[int]: Indices of varying parameters in the 14-element physics array.

property fixed_indices: list[int]: Indices of fixed parameters in the 14-element physics array.

get_initial_values()[source]

Get initial parameter values for optimization.

Returns the config-specified starting point, not the current fitted state. Reads from the frozen snapshot set at construction time so that repeated calls (e.g. across multi-angle loops) always return the same config values even after model.set_params() has mutated space.values.

Return type:: ndarray
Returns:: Array of shape (n_varying,) with initial values for varying params.

get_full_values()[source]

Get all 14 parameter values.

Returns a read-only cached array (writeable=False). Use .copy() if mutation is required.

Return type:: ndarray
Returns:: Array of shape (14,).

get_bounds()[source]

Get bounds for varying physics parameters.

Return type:: tuple[ndarray, ndarray]
Returns:: (lower, upper) each of shape (n_varying,).

expand_varying_to_full(varying_params)[source]

Expand varying parameters to full 14-parameter array.

Fixed parameters are filled from stored values.

Parameters:: varying_params (ndarray | Array) – Array of shape (n_varying,).
Return type:: ndarray
Returns:: Array of shape (14,).

extract_varying(full_params)[source]

Extract varying parameters from full array.

Parameters:: full_params (ndarray | Array) – Array of shape (14,).
Return type:: ndarray
Returns:: Array of shape (n_varying,).

update_values(params)[source]

Update stored parameter values.

Parameters:: params (ndarray | dict[str, float]) – Either array of shape (14,) or dict with param names.
Return type:: None

get_parameter_dict()[source]

Get current parameter values as dictionary.

Return type:: dict[str, float]

set_vary(name, vary)[source]

Set whether a parameter varies in optimization.

Invalidates relevant caches.

Parameters:

name (str) – Parameter name (physics or scaling).
vary (bool) – Whether to vary this parameter.

Return type:

None

set_bounds(name, lower, upper)[source]

Set bounds for a parameter.

Invalidates the bounds cache for any query that includes this parameter.

Parameters:

name (str) – Parameter name (physics or scaling).
lower (float) – Lower bound.
upper (float) – Upper bound.

Return type:

None

validate_physics(params=None)[source]

Validate parameters against physics constraints.

Parameters:: params (ndarray | None) – Full parameter array of shape (14,), or None to use stored values.
Return type:: list[str]
Returns:: List of violation messages (empty if valid).

classmethod from_config(config)[source]

Create ParameterManager from configuration dictionary.

Parameters:: config (dict[str, Any]) – Full configuration dict.
Return type:: ParameterManager
Returns:: Configured ParameterManager.

get_group_values(group)[source]

Get parameter values for a specific group.

Parameters:: group (str) – Group name (‘reference’, ‘sample’, ‘velocity’, ‘fraction’, ‘angle’, ‘scaling’).
Return type:: dict[str, float]
Returns:: Dict mapping parameter names to values.

get_parameter_bounds(parameter_names=None)[source]

Get parameter bounds configuration with caching.

Parameters:: parameter_names (list[str] | None) – Names of parameters to retrieve bounds for. If None, returns bounds for all 16 parameters (14 physics + 2 scaling) in canonical order.
Return type:: list[BoundDict]
Returns:: List of BoundDict entries with keys ‘name’, ‘min’, ‘max’, ‘type’.

Notes

Results are cached per unique (sorted) parameter set. Cache is invalidated automatically by set_bounds().

get_bounds_as_tuples(parameter_names=None)[source]

Get parameter bounds as a list of (min, max) tuples.

Convenience method for compatibility with optimization code that expects the scipy-style bounds format.

Parameters:: parameter_names (list[str] | None) – Parameter names. If None, uses all 16 parameters.
Return type:: list[tuple[float, float]]
Returns:: List of (min, max) tuples, one per parameter.

get_bounds_as_arrays(parameter_names=None)[source]

Get parameter bounds as separate lower and upper numpy arrays.

Convenience method for NLSQ and JAX optimizers that consume separate lower/upper bound arrays.

Parameters:: parameter_names (list[str] | None) – Parameter names. If None, uses all 16 parameters.
Return type:: tuple[ndarray, ndarray]
Returns:: (lower_bounds, upper_bounds) as numpy arrays of shape (n_params,).

get_active_parameters()[source]

Get physics parameter names that are marked as varying.

Returns the 14-element physics parameters (excludes scaling) whose vary flag is True in the current ParameterSpace. Falls back to all 14 physics parameters if the space has no explicit vary flags set.

Results are cached; call set_vary() to invalidate automatically.

Return type:: list[str]
Returns:: List of varying physics parameter names in canonical order.

get_all_parameter_names()[source]

Get all parameter names: scaling parameters first, then physics.

Return type:: list[str]
Returns:: List of 16 names (contrast, offset, then the 14 physics params) in canonical order.

get_effective_parameter_count()[source]

Number of active (varying) physics parameters, excluding scaling.

Return type:: int
Returns:: Count of physics parameters whose vary flag is True.

get_total_parameter_count()[source]

Total parameter count including both scaling and physics parameters.

Return type:: int
Returns:: Always 16 for the heterodyne model (14 physics + 2 scaling).

get_fixed_parameters()[source]

Return physics parameters that are held fixed during optimization.

A parameter is considered fixed when its vary flag is False in the ParameterSpace. Scaling parameters (contrast, offset) are excluded from this result — use get_parameter_dict() to access their values.

Return type:: dict[str, float]
Returns:: Dict mapping fixed physics parameter name to its current value.

is_parameter_active(param_name)[source]

Check whether a physics parameter is active (vary=True).

Parameters:: param_name (str) – Physics parameter name to check. Must be one of the 14 physics parameters; scaling names always return False.
Return type:: bool
Returns:: True if the parameter’s vary flag is True, False otherwise.

get_optimizable_parameters()[source]

Return physics parameters that should be optimized.

Equivalent to active parameters (vary=True). Scaling parameters are handled separately and are not included.

Return type:: list[str]
Returns:: List of physics parameter names with vary=True, in canonical order.

validate_physical_constraints(params=None, severity_level='warning')[source]

Validate physics-based constraints beyond simple bound checking.

Checks for physically impossible or unusual parameter combinations based on the heterodyne two-component scattering model.

Parameters:

params (dict[str, float] | ndarray | None) – Parameter dict, array of shape (14,), or None to use stored values. Dict keys must be physics parameter names.
severity_level (str) – Minimum severity to include in the result. One of: - "error" — physically impossible values only. - "warning" — unusual but possible values (default). - "info" — all noteworthy observations. Currently the heterodyne validator does not distinguish severity internally; this argument is accepted for API parity with homodyne and is reserved for future use.

Return type:

ValidationResult

Returns:

ValidationResult with is_valid, errors, and warnings.

__repr__()[source]

Concise string representation of manager state.

Return type:: str

__init__(space=<factory>)

Parameter Space¶

Parameter space definition with prior distributions for Bayesian inference.

class heterodyne.config.parameter_space.PriorType[source]

Bases: Enum

Available prior distribution types.

UNIFORM = 'uniform'

NORMAL = 'normal'

TRUNCATED_NORMAL = 'truncated_normal'

LOGNORMAL = 'lognormal'

HALFNORMAL = 'halfnormal'

EXPONENTIAL = 'exponential'

BETA_SCALED = 'beta_scaled'

class heterodyne.config.parameter_space.PriorDistribution[source]

Bases: object

Prior distribution specification for a parameter.

prior_type: PriorType

params: dict[str, float]

classmethod uniform(low, high)[source]

Create uniform prior.

Return type:: PriorDistribution

classmethod normal(loc, scale)[source]

Create normal (Gaussian) prior.

Return type:: PriorDistribution

classmethod lognormal(loc, scale)[source]

Create log-normal prior (for positive parameters).

Return type:: PriorDistribution

classmethod halfnormal(scale)[source]

Create half-normal prior (for positive parameters).

Return type:: PriorDistribution

classmethod truncated_normal(loc, scale, low, high)[source]

Create truncated normal prior (bounded Gaussian).

Return type:: PriorDistribution

classmethod beta_scaled(low, high, concentration1, concentration2)[source]

Create a Beta prior scaled to [low, high].

The distribution is Beta(concentration1, concentration2) affine-transformed to the interval [low, high]. This is useful for bounded parameters where you want to express a prior belief about the shape within the bounds.

Parameters:

low (float) – Lower bound of the support.
high (float) – Upper bound of the support.
concentration1 (float) – First concentration parameter (alpha > 0).
concentration2 (float) – Second concentration parameter (beta > 0).

Return type:

PriorDistribution

Returns:

PriorDistribution with BETA_SCALED type.

to_numpyro(name)[source]

Convert to NumPyro distribution.

Parameters:: name (str) – Parameter name for the distribution
Return type:: Any
Returns:: NumPyro distribution object

__init__(prior_type, params=<factory>)

class heterodyne.config.parameter_space.ParameterSpace[source]

Bases: object

Complete parameter space for heterodyne model optimization.

Manages parameter values, bounds, vary flags, and priors.

values: dict[str, float]

vary: dict[str, bool]

bounds: dict[str, tuple[float, float]]

priors: dict[str, PriorDistribution]

__post_init__()[source]

Initialize with defaults from registry.

Return type:: None

property n_total: int: Total number of parameters.

property n_varying: int: Number of parameters that vary in optimization.

property varying_names: list[str]: Names of parameters that vary (physics + scaling).

property fixed_names: list[str]: Names of parameters that are fixed.

property varying_physics_names: list[str]: Names of varying physics parameters (excludes scaling).

property scaling_values: dict[str, float]: Get contrast and offset values.

get_initial_array()[source]

Get initial values as numpy array in canonical order.

Return type:: ndarray
Returns:: Array of shape (14,) with parameter values

to_config()[source]

Serialize this space to a dict compatible with from_config().

Produces the initial_parameters flat-format understood by _apply_initial_parameters(). Bounds and priors are not serialized — workers rebuild them from the registry defaults. Only values and active_parameters (vary flags) are round-tripped.

Return type:: dict[str, Any]
Returns:: Config dict that from_config() can reconstruct into an equivalent ParameterSpace (same values and varying_names).

get_bounds_arrays()[source]

Get bounds as numpy arrays.

Return type:: tuple[ndarray, ndarray]
Returns:: (lower_bounds, upper_bounds) each of shape (14,)

get_vary_mask()[source]

Get boolean mask for varying parameters.

Return type:: ndarray
Returns:: Boolean array of shape (14,)

array_to_dict(arr)[source]

Convert parameter array to dictionary.

Parameters:: arr (ndarray | Array) – Array of shape (14,)
Return type:: dict[str, float]
Returns:: Dict mapping parameter names to values

update_from_dict(params)[source]

Update parameter values from dictionary.

Parameters:: params (dict[str, float]) – Dict with parameter names as keys
Raises:: ValueError – If a key doesn’t match any known parameter
Return type:: None

validate()[source]

Validate parameter space configuration.

Covers all 16 parameters (14 physics + 2 scaling) so a malformed contrast/offset bound is caught at config-load time instead of propagating into the optimizer.

Bounds are checked for every parameter — varying and fixed — because fixed parameters still drive model outputs and CMC warm-starts. The one explicit exception: parameters in _FIXED_ZERO_SENTINELS (currently v0) are allowed to be 0.0 when fixed, because their min_bound is an optimizer stability floor that does not apply when the optimizer never touches the value.

Return type:: list[str]
Returns:: List of validation error messages (empty if valid)

convert_to_beta_priors()[source]

Convert all TruncatedNormal priors to BetaScaled priors.

For each parameter whose prior is TRUNCATED_NORMAL, this method computes equivalent Beta concentration parameters via the method of moments and replaces the prior in-place with a BETA_SCALED distribution over the same bounds.

Parameters with other prior types are left unchanged.

Return type:: None

with_single_angle_stabilization()[source]

Return a new ParameterSpace with tightened bounds for single-angle analysis.

Narrows contrast bounds to [value-0.2, value+0.2] and offset bounds to [value-0.1, value+0.1], clamped to the original bounds.

Return type:: ParameterSpace
Returns:: A new ParameterSpace with tightened scaling bounds.

classmethod from_config(config)[source]

Create ParameterSpace from configuration dictionary.

Supports two input formats (homodyne parity):

Grouped format (preferred) — parameters.{group}.{param}:

parameters:
  reference:
    D0_ref:
      value: 5000.0
      min: 200.0
      max: 50000.0
      vary: true

Flat format — initial_parameters.parameter_names + values:

initial_parameters:
  parameter_names: [D0_ref, alpha_ref]
  values: [5000.0, 0.5]
  active_parameters: [D0_ref]   # optional vary subset

When both are present, grouped format takes precedence (it is applied second so its values overwrite flat-format values).

Parameters:: config (dict[str, Any]) – Config dict with ‘parameters’ and/or ‘initial_parameters’ sections.
Return type:: ParameterSpace
Returns:: Configured ParameterSpace

__init__(values=<factory>, vary=<factory>, bounds=<factory>, priors=<factory>)

heterodyne.config.parameter_space.clamp_to_open_interval(value, low, high, epsilon=1e-6)[source]

Clamp value to the open interval (low+epsilon, high-epsilon).

Useful for Beta distribution parameters that must be strictly within their support bounds.

Parameters:

value (float) – Value to clamp.
low (float) – Lower bound of the closed interval.
high (float) – Upper bound of the closed interval.
epsilon (float) – Margin to inset from the bounds.

Return type:

float

Returns:

Clamped value in (low+epsilon, high-epsilon).

Parameter Names¶

Parameter name constants for 14-parameter heterodyne model.

The heterodyne model describes two-component correlation with: - Reference component: diffusive transport (D0_ref, alpha_ref, D_offset_ref) - Sample component: diffusive transport (D0_sample, alpha_sample, D_offset_sample) - Velocity field: time-dependent flow (v0, beta, v_offset) - Fraction: sample fraction evolution (f0, f1, f2, f3) - Angle: flow angle relative to q-vector (phi0)

heterodyne.config.parameter_names.get_param_index(name)[source]

Get index of parameter in flattened array.

Parameters:: name (str) – Parameter name
Return type:: int
Returns:: Index (0-13)
Raises:: KeyError – If parameter name is invalid

heterodyne.config.parameter_names.get_group_indices(group)[source]

Get indices for all parameters in a group.

Parameters:: group (str) – Group name (‘reference’, ‘sample’, ‘velocity’, ‘fraction’, ‘angle’)
Return type:: tuple[int, ...]
Returns:: Tuple of indices

Type Aliases¶

TypedDict definitions for configuration structures.

class heterodyne.config.types.ParameterConfig[source]

Bases: TypedDict

Configuration for a single parameter.

value: float

min: float

max: float

vary: bool

prior: NotRequired[str]

prior_params: NotRequired[dict[str, float]]

class heterodyne.config.types.ParameterGroupConfig[source]

Bases: TypedDict

Configuration for a parameter group.

Keys use full parameter names matching parameter_space.from_config(). Group ‘reference’: D0_ref, alpha_ref, D_offset_ref Group ‘sample’: D0_sample, alpha_sample, D_offset_sample Group ‘velocity’: v0, beta, v_offset Group ‘fraction’: f0, f1, f2, f3 Group ‘angle’: phi0