heterodyne.core.config

Configuration Management for Heterodyne Scattering Analysis

Centralized configuration system for XPCS analysis under nonequilibrium conditions. Provides JSON-based configuration management with validation, hierarchical parameter organization, and performance optimization features.

Key Features: - Hierarchical JSON configuration with validation - Runtime parameter override capabilities - Performance-optimized configuration access with caching - Comprehensive logging system with rotation and formatting - Physical parameter validation and bounds checking - Angle filtering configuration for computational efficiency - Test configuration management for different analysis scenarios

Configuration Structure: - analyzer_parameters: Core physics parameters (q-vector, time steps, geometry) - experimental_data: Data paths, file formats, and loading options - analysis_settings: Mode selection (static vs laminar flow) - optimization_config: Method settings, hyperparameters, angle filtering - parameter_space: Physical bounds, priors, and parameter constraints - performance_settings: Computational optimization flags

Authors: Wei Chen, Hongrui He Institution: Argonne National Laboratory

class heterodyne.core.config.LoggingConfig

Bases: TypedDict

Typed configuration for logging system.

log_to_file: bool

log_to_console: bool

log_filename: str

level: str

format: str

rotation: dict[str, int | str]

class heterodyne.core.config.AngleRange

Bases: TypedDict

Typed configuration for angle filtering ranges.

min_angle: float

max_angle: float

class heterodyne.core.config.AngleFilteringConfig

Bases: TypedDict

Typed configuration for angle filtering.

enabled: bool

target_ranges: list[AngleRange]

fallback_to_all_angles: bool

class heterodyne.core.config.OptimizationMethodConfig

Bases: TypedDict

Typed configuration for optimization method parameters.

maxiter: int

xatol: float

fatol: float

class heterodyne.core.config.ClassicalOptimizationConfig

Bases: TypedDict

Typed configuration for classical optimization methods.

methods: list[str]

method_options: dict[str, OptimizationMethodConfig]

class heterodyne.core.config.OptimizationConfig

Bases: TypedDict

Typed configuration for optimization settings.

angle_filtering: AngleFilteringConfig

classical_optimization: ClassicalOptimizationConfig

class heterodyne.core.config.ParameterBound

Bases: TypedDict

Typed configuration for parameter bounds.

name: str

min: float

max: float

type: str

class heterodyne.core.config.ParameterSpaceConfig

Bases: TypedDict

Typed configuration for parameter space definition.

bounds: list[ParameterBound]

class heterodyne.core.config.InitialParametersConfig

Bases: TypedDict

Typed configuration for initial parameter values.

parameter_names: list[str]

active_parameters: NotRequired[list[str]]

class heterodyne.core.config.AnalysisSettings

Bases: TypedDict

Typed configuration for analysis mode settings.

static_mode: bool

static_submode: NotRequired[str]

model_description: str

class heterodyne.core.config.ExperimentalDataConfig

Bases: TypedDict

Typed configuration for experimental data paths.

data_folder_path: str

data_file_name: str

phi_angles_path: str

phi_angles_file: str

exchange_key: str

cache_file_path: str

cache_filename_template: str

heterodyne.core.config.configure_logging(cfg=None, *, level=None, log_file=None)

Configure centralized logging system with hierarchy and handlers.

This function sets up a complete logging infrastructure: - Creates a logger hierarchy (root + module logger) - Sets up RotatingFileHandler with size-based rotation - Optionally creates StreamHandler for console output - Applies consistent formatting and log levels

Parameters:

• cfg (dict, optional) – Logging configuration dictionary with keys: - log_to_file: bool, enable file logging - log_to_console: bool, enable console logging - log_filename: str, log file path - level: str, logging level (‘DEBUG’, ‘INFO’, ‘WARNING’, ‘ERROR’) - format: str, log message format string - rotation: dict with ‘max_bytes’ and ‘backup_count’

• level (Any, optional) – Logging level for backward compatibility (can be string or logging constant)

• log_file (str, optional) – Log file path for backward compatibility

Returns:

Configured logger instance for reuse

Return type:

logging.Logger

class heterodyne.core.config.ConfigManager(config_file='heterodyne_config.json', config=None)

Bases: object

Centralized configuration manager for heterodyne scattering analysis.

This class orchestrates the entire configuration system for XPCS analysis, providing structured access to all analysis parameters with validation, caching, and runtime override capabilities.

Core Responsibilities: - JSON configuration file loading with comprehensive error handling - Hierarchical parameter validation (physics, computation, file paths) - Performance-optimized configuration access through intelligent caching - Runtime configuration overrides for analysis mode switching - Logging system setup with rotation and appropriate formatting - Test configuration management for different experimental scenarios

Configuration Hierarchy: - analyzer_parameters: Physics parameters (q-vector, time steps, gap size) - experimental_data: Data file paths, loading options, caching settings - analysis_settings: Mode selection (static/laminar flow), model descriptions - optimization_config: Method settings, angle filtering, hyperparameters - parameter_space: Physical parameter bounds, prior distributions - performance_settings: Parallelization, computational optimizations - validation_rules: Data quality checks and minimum requirements - advanced_settings: Fine-tuning options for specialized use cases

Usage:

config_manager = ConfigManager(‘my_config.json’) is_static = config_manager.is_static_mode_enabled() angle_ranges = config_manager.get_target_angle_ranges()

Parameters:

• config_file (str)

• config (dict[str, Any] | None)

__init__(config_file='heterodyne_config.json', config=None)

Initialize configuration manager.

Parameters:

• config_file (str) – Path to JSON configuration file

• config (dict, optional) – Configuration dictionary (if provided, config_file is ignored)

config: dict[str, Any] | None

load_config(config_file=None)

Load and parse JSON configuration file with comprehensive error handling and security validation.

Implements performance-optimized loading with buffering, structure optimization for runtime access, security validation, and graceful fallback to default configuration if primary config fails.

Security Features: - Input validation and sanitization - Path traversal prevention - Configuration structure validation - Parameter bounds checking

Error Handling: - FileNotFoundError: Missing configuration file - JSONDecodeError: Malformed JSON syntax - ValidationError: Security validation failures - General exceptions: Unexpected loading issues

Performance Optimizations: - 8KB buffering for efficient file I/O - Configuration structure caching for fast access - Timing instrumentation for performance monitoring

Parameters:

config_file (str | None)

Return type:

dict[str, Any] | None

validate_config()

Comprehensive validation of configuration parameters.

Performs multi-level validation to ensure configuration integrity:

Structural Validation: - Required sections presence (analyzer_parameters, experimental_data, etc.) - Configuration hierarchy completeness - Parameter type consistency

Physical Parameter Validation: - Frame range consistency (start < end, sufficient frames) - Wavevector positivity and reasonable magnitude - Time step positivity - Gap size physical reasonableness

Data Validation: - Minimum frame count requirements - Parameter bounds consistency - File path accessibility (optional)

Raises:

• ValueError – Invalid configuration parameters or structure

• FileNotFoundError – Missing required data files (if validation enabled)

Return type:

bool

validate_parameter_bounds(parameters)

Validate that parameters are within the configured bounds.

Parameters:

parameters (dict[str, float]) – Dictionary of parameter names and values to validate

Returns:

True if all parameters are within bounds, False otherwise

Return type:

bool

get_parameter(section, parameter, default=None)

Get a parameter value from a specific configuration section.

Parameters:

• section (str) – Configuration section name

• parameter (str) – Parameter name within the section

• default (Any, optional) – Default value to return if parameter not found

Returns:

Parameter value

Return type:

Any

Raises:

KeyError – If parameter not found and no default provided

set_parameter(section, parameter, value)

Set a parameter value in a specific configuration section.

Parameters:

• section (str) – Configuration section name

• parameter (str) – Parameter name within the section

• value (Any) – Value to set for the parameter

Raises:

KeyError – If section not found in configuration

Return type:

None

merge_configs(update_config)

Merge an update configuration with the current configuration.

Parameters:

update_config (dict[str, Any]) – Configuration dictionary to merge with current config

Returns:

Merged configuration dictionary

Return type:

dict[str, Any]

save_config(file_path)

Save the current configuration to a JSON file.

Parameters:

file_path (str) – Path where the configuration file should be saved

Raises:

• ValueError – If no configuration is loaded

• FileNotFoundError – If the parent directory doesn’t exist

Return type:

None

setup_logging()

Configure logging based on configuration using centralized configure_logging().

Return type:

Logger | None

get(*keys, default=None)

Get nested configuration value.

Parameters:

• *keys (str) – Sequence of nested keys

• default (any) – Default value if key not found

Return type:

Configuration value or default

get_angle_filtering_config()

Get angle filtering configuration with defaults.

Returns:

Angle filtering configuration including: - enabled: bool, whether angle filtering is enabled - target_ranges: list of dicts with min_angle and max_angle - fallback_to_all_angles: bool, whether to use all angles if no targets found

Return type:

dict

is_angle_filtering_enabled()

Check if angle filtering is enabled in configuration.

Returns:

True if angle filtering should be used, False otherwise

Return type:

bool

get_target_angle_ranges()

Get list of target angle ranges for optimization.

Returns:

List of (min_angle, max_angle) tuples in degrees

Return type:

list of tuple

should_fallback_to_all_angles()

Check if system should fallback to all angles when no targets found.

Returns:

True if should fallback to all angles, False to raise error

Return type:

bool

is_static_mode_enabled()

Check if static mode is enabled in configuration.

DEPRECATED: Static mode has been removed in favor of the heterodyne model. This method now raises an error if static mode is detected.

Returns:

Always returns False (static mode no longer supported)

Return type:

bool

Raises:

ValueError – If static mode configuration is detected

get_static_submode()

Get the static sub-mode for analysis.

DEPRECATED: Static submodes have been removed.

Returns:

Always returns None (static modes no longer supported)

Return type:

str | None

Raises:

ValueError – If static submode configuration is detected

get_analysis_mode()

Get the current analysis mode.

Returns:

“heterodyne” - 14-parameter model

Return type:

str

get_active_parameters()

Get list of active parameters from configuration.

Returns:

List of parameter names for the 14-parameter heterodyne model. Always returns all 14 parameter names.

Return type:

list[str]

get_effective_parameter_count()

Get the effective number of model parameters.

Returns:

Always returns 14 for the heterodyne model. The heterodyne model uses 14 parameters: - Reference transport coefficients (3): D0_ref, alpha_ref, D_offset_ref - Sample transport coefficients (3): D0_sample, alpha_sample, D_offset_sample - Velocity coefficients (3): v0, beta, v_offset - Fraction coefficients (4): f0, f1, f2, f3 - Flow angle (1): phi0

Return type:

int

get_parameter_metadata()

Get metadata for all 14 heterodyne parameters.

Returns:

Parameter metadata with units and descriptions

Return type:

dict[str, dict[str, str]]

get_parameter_bounds()

Get recommended bounds for all 14 heterodyne parameters.

Returns:

List of (min, max) bounds for each parameter

Return type:

list[tuple[float, float]]

get_default_14_parameters()

Get default values for 14-parameter heterodyne model.

For backward compatibility, initializes sample parameters to match reference parameters (g1_sample = g1_ref initially).

Returns:

Default parameter values

Return type:

list[float]

list_available_templates()

List all available configuration templates.

Returns:

List of available template names

Return type:

list[str]

load_template(template_name)

Load a configuration template by name.

Parameters:

template_name (str) – Name of the template to load

Returns:

Template configuration dictionary

Return type:

dict[str, Any]

Raises:

• ValueError – If template name is not found

• FileNotFoundError – If template file doesn’t exist

resolve_environment_variables(config)

Resolve environment variables in configuration.

Parameters:

config (dict[str, Any]) – Configuration dictionary that may contain environment variable references

Returns:

Configuration with environment variables substituted

Return type:

dict[str, Any]

create_backup()

Create a backup of the current configuration.

Returns:

Deep copy of the current configuration

Return type:

dict[str, Any]

Raises:

ValueError – If no configuration is loaded

restore_from_backup(backup)

Restore configuration from a backup.

Parameters:

backup (dict[str, Any]) – Configuration backup to restore

Return type:

None

get_config_differences(other_config)

Get differences between current configuration and another configuration.

Parameters:

other_config (dict[str, Any]) – Configuration to compare against

Returns:

Dictionary containing differences

Return type:

dict[str, Any]

get_analysis_settings()

Get analysis settings with defaults.

Returns:

Analysis settings including static_mode flag and descriptions

Return type:

dict[str, Any]

class heterodyne.core.config.PerformanceMonitor

Bases: object

Performance monitoring and profiling utilities.

Provides lightweight profiling and memory monitoring for optimization of computational kernels.

__init__()

Return type:

None

timings: dict[str, list[float]]

memory_usage: dict[str, float]

__call__(operation_name)

Make PerformanceMonitor callable as a context manager.

Parameters:

operation_name (str) – Name of the operation being monitored

Returns:

Context manager for timing the operation

Return type:

_TimingContext

start(operation_name='default_operation')

Start monitoring an operation.

Parameters:

operation_name (str, optional) – Name of the operation being monitored

Returns:

Context manager for timing the operation

Return type:

_TimingContext

time_function(func_name)

Context manager for timing function execution.

Parameters:

• func_name (str) – Name of function being timed

• Usage

• -----

• monitor.time_function("my_function") (with) – # function code here pass

Return type:

_TimingContext

get_timing_summary()

Get summary statistics for all timed functions.

Return type:

dict[str, dict[str, float]]

reset_timings()

Clear all timing data.

Return type:

None